diff --git a/apps/launcher/package.json b/apps/launcher/package.json index 1ecfc1f..3d1e184 100644 --- a/apps/launcher/package.json +++ b/apps/launcher/package.json @@ -9,6 +9,7 @@ "@codex-git/host-adapter": "*", "@codex-git/host-adapter-codex-cdp": "*", "@codex-git/host-adapter-standalone": "*", + "@codex-git/repository-engine": "*", "@codex-git/server": "*" }, "devDependencies": { diff --git a/apps/launcher/src/standalone-runtime.ts b/apps/launcher/src/standalone-runtime.ts index 876a27a..7865e16 100644 --- a/apps/launcher/src/standalone-runtime.ts +++ b/apps/launcher/src/standalone-runtime.ts @@ -2,6 +2,11 @@ import type { Server } from 'node:http'; import { fileURLToPath } from 'node:url'; import type { HostConnection } from '@codex-git/host-adapter'; +import { + createRepositoryEngine, + type RepositorySession, +} from '@codex-git/repository-engine'; +import type { AbsolutePath, RepositoryId } from '@codex-git/protocol'; import { startLoopbackServer, type LoopbackServer } from '@codex-git/server'; import { StandaloneHostAdapter } from '@codex-git/host-adapter-standalone'; import { createServer as createViteServer, type ViteDevServer } from 'vite'; @@ -14,6 +19,7 @@ const uiConfigPath = fileURLToPath( ); export interface StandaloneRuntimeOptions { + readonly projectPath?: string; readonly surfacePort?: number; } @@ -30,17 +36,34 @@ export async function startStandaloneRuntime( let protocolServer: LoopbackServer | undefined; let surfaceServer: ViteDevServer | undefined; let hostConnection: HostConnection | null = null; + let repositorySession: RepositorySession | undefined; + let invalidationPump = Promise.resolve(); async function closeResources(): Promise { await Promise.all([ hostConnection?.close(), + repositorySession?.close(), surfaceServer?.close(), protocolServer?.close(), ]); + await invalidationPump; } try { protocolServer = await startLoopbackServer({ allowedOrigins: ['null'] }); + if (options.projectPath !== undefined) { + repositorySession = await createRepositoryEngine().open( + options.projectPath as AbsolutePath, + ); + const opened = await repositorySession.requestRefresh(); + if (opened.kind === 'repository') { + invalidationPump = forwardRepositoryInvalidations( + repositorySession, + protocolServer, + opened.repository.repositoryId, + ); + } + } surfaceServer = await createViteServer({ configFile: uiConfigPath, plugins: [protocolBootstrapPlugin(protocolServer.sessionUrl)], @@ -81,6 +104,29 @@ export async function startStandaloneRuntime( } } +async function forwardRepositoryInvalidations( + session: RepositorySession, + server: Pick, + repositoryId: RepositoryId, +): Promise { + for await (const invalidation of session.subscribe()) { + server.publish( + invalidation.kind === 'operation' + ? { + kind: 'operation_progress', + operationId: invalidation.operation.operationId, + phase: invalidation.operation.phase, + progress: invalidation.operation.progress, + } + : { + kind: 'repository_revision', + repositoryId, + repositoryRevision: invalidation.repositoryRevision, + }, + ); + } +} + function serverUrl( server: Pick | null, pathname: string, diff --git a/package-lock.json b/package-lock.json index b4f1e18..ef7b4be 100644 --- a/package-lock.json +++ b/package-lock.json @@ -41,6 +41,7 @@ "@codex-git/host-adapter": "*", "@codex-git/host-adapter-codex-cdp": "*", "@codex-git/host-adapter-standalone": "*", + "@codex-git/repository-engine": "*", "@codex-git/server": "*" }, "devDependencies": { diff --git a/packages/repository-engine/src/index.ts b/packages/repository-engine/src/index.ts index f0c08e2..74ac802 100644 --- a/packages/repository-engine/src/index.ts +++ b/packages/repository-engine/src/index.ts @@ -10,6 +10,7 @@ export { export { type IndexSnapshot, type RefSnapshot, + type UpstreamSnapshot, type WorktreeObservationError, type WorktreeStatusSummary, } from './repository-observation.js'; @@ -20,8 +21,8 @@ export { type RefreshState, type RepositoryInvalidation, type RepositoryOpenResult, - type RepositorySession, RepositorySessionFailure, type RepositorySnapshot, type WorktreeFreshness, } from './repository-publication.js'; +export { type RepositorySession } from './repository-session.js'; diff --git a/packages/repository-engine/src/observation-publication.ts b/packages/repository-engine/src/observation-publication.ts index 55d6878..33032e5 100644 --- a/packages/repository-engine/src/observation-publication.ts +++ b/packages/repository-engine/src/observation-publication.ts @@ -6,6 +6,7 @@ import type { IndexSnapshot, RefSnapshot, RepositoryObservation, + UpstreamSnapshot, WorktreeObservation, WorktreeObservationError, WorktreeStatusSummary, @@ -26,6 +27,7 @@ export interface PublishedObservationWorktree extends Omit< readonly freshness: WorktreeFreshness; readonly index: IndexSnapshot | null; readonly status: WorktreeStatusSummary | null; + readonly upstream: UpstreamSnapshot; } export type WorktreeFreshness = @@ -35,21 +37,26 @@ export type WorktreeFreshness = | { readonly kind: 'failed'; readonly error: WorktreeObservationError }; export interface PublishedObservationResult extends PublishedRepositoryObservation { - readonly privateRefsEvidence: string; + readonly privateRefsEvidence: PrivateRefsEvidence; readonly refsChanged: boolean; readonly worktreeChanged: boolean; } +export interface PrivateRefsEvidence { + readonly shared: string; + readonly upstreams: string; +} + export function publishObservedFacts( discovery: RepositoryDiscovery, previous?: PublishedRepositoryObservation, observation?: RepositoryObservation, - previousPrivateRefsEvidence?: string, + previousPrivateRefsEvidence?: PrivateRefsEvidence, ): PublishedObservationResult { const shared = observation?.shared ?? { refs: previous?.refs ?? [], remotes: previous?.remotes ?? [], - privateRefsEvidence: previousPrivateRefsEvidence ?? '', + privateRefsEvidence: previousPrivateRefsEvidence?.shared ?? '', }; const previousWorktrees = new Map( previous?.worktrees.map((worktree) => [worktree.worktreeId, worktree]), @@ -57,14 +64,26 @@ export function publishObservedFacts( const observations = new Map( observation?.worktrees.map((worktree) => [worktree.worktreeId, worktree]), ); + const sharedRefsChanged = + observation !== undefined && + shared.privateRefsEvidence !== previousPrivateRefsEvidence?.shared; let worktreeChanged = previous === undefined; const worktrees = discovery.worktrees.map((worktree) => { const prior = previousWorktrees.get(worktree.worktreeId); const observed = observations.get(worktree.worktreeId); - if (observation !== undefined && observed === undefined) { + if ( + observation !== undefined && + observation.complete !== false && + observed === undefined + ) { throw new Error('Repository observation omitted a registered Worktree.'); } - const observedFacts = publishWorktreeObservation(worktree, observed, prior); + const observedFacts = + observation?.complete === false && + observed === undefined && + prior !== undefined + ? retainWorktreeObservation(prior, sharedRefsChanged) + : publishWorktreeObservation(worktree, observed, prior); const published: Omit = { worktreeId: worktree.worktreeId, generation: worktree.generation, @@ -77,6 +96,7 @@ export function publishObservedFacts( freshness: observedFacts.freshness, index: observedFacts.index, status: observedFacts.status, + upstream: observedFacts.upstream, }; const changed = prior === undefined || @@ -89,26 +109,68 @@ export function publishObservedFacts( }; }); worktreeChanged ||= previousWorktrees.size !== worktrees.length; + const privateRefsEvidence: PrivateRefsEvidence = + observation === undefined + ? (previousPrivateRefsEvidence ?? { shared: '', upstreams: '[]' }) + : { + shared: shared.privateRefsEvidence, + upstreams: JSON.stringify( + worktrees.map(({ worktreeId, upstream }) => ({ + worktreeId, + upstream, + })), + ), + }; + const upstreamChanged = worktrees.some((worktree) => { + const prior = previousWorktrees.get(worktree.worktreeId); + return ( + prior !== undefined && + JSON.stringify(worktree.upstream) !== JSON.stringify(prior.upstream) + ); + }); return { refs: shared.refs, remotes: shared.remotes, worktrees, - privateRefsEvidence: shared.privateRefsEvidence, + privateRefsEvidence, refsChanged: previous === undefined || - (observation !== undefined && - shared.privateRefsEvidence !== previousPrivateRefsEvidence), + (observation !== undefined && (sharedRefsChanged || upstreamChanged)), worktreeChanged, }; } +function retainWorktreeObservation( + previous: PublishedObservationWorktree, + sharedRefsChanged: boolean, +): Pick< + PublishedObservationWorktree, + 'freshness' | 'head' | 'index' | 'status' | 'upstream' +> { + return { + freshness: sharedRefsChanged + ? { + kind: 'stale', + error: { + code: 'not_observed', + message: 'Shared refs changed before this Worktree was observed.', + }, + } + : previous.freshness, + head: previous.head, + index: previous.index, + status: previous.status, + upstream: previous.upstream, + }; +} + function publishWorktreeObservation( worktree: DiscoveredWorktree, observed: WorktreeObservation | undefined, previous: PublishedObservationWorktree | undefined, ): Pick< PublishedObservationWorktree, - 'freshness' | 'head' | 'index' | 'status' + 'freshness' | 'head' | 'index' | 'status' | 'upstream' > { if (observed?.kind === 'fresh') { return { @@ -116,6 +178,7 @@ function publishWorktreeObservation( head: observed.head, index: observed.index, status: observed.status, + upstream: observed.upstream, }; } if ( @@ -127,6 +190,7 @@ function publishWorktreeObservation( head: worktree.head, index: null, status: null, + upstream: previous?.upstream ?? { kind: 'unavailable' }, }; } if (observed?.kind === 'failed') { @@ -140,6 +204,7 @@ function publishWorktreeObservation( head: previous.head, index: previous.index, status: previous.status, + upstream: previous.upstream, }; } return { @@ -147,6 +212,7 @@ function publishWorktreeObservation( head: worktree.head, index: null, status: null, + upstream: { kind: 'unavailable' }, }; } return { @@ -154,6 +220,10 @@ function publishWorktreeObservation( head: worktree.head, index: null, status: null, + upstream: + worktree.head.kind === 'detached' + ? { kind: 'not_applicable', reason: 'detached_head' } + : { kind: 'unavailable' }, }; } diff --git a/packages/repository-engine/src/operation-coordinator-lifecycle.ts b/packages/repository-engine/src/operation-coordinator-lifecycle.ts index a51ab71..3f0f7a6 100644 --- a/packages/repository-engine/src/operation-coordinator-lifecycle.ts +++ b/packages/repository-engine/src/operation-coordinator-lifecycle.ts @@ -2,25 +2,21 @@ import type { OperationId, OperationResult } from '@codex-git/protocol'; import { createOperationLifecycleStore, + systemTimeoutScheduler, + validateTimeoutMilliseconds, type LifecycleCloseResult, type ManagedOperationRecord, type OperationLifecycleStore, + type TimeoutScheduler, } from './operation-lifecycle.js'; -interface TimeoutScheduler { - schedule( - milliseconds: number, - onTimeout: () => void, - ): { - cancel(): void; - }; -} - export interface CoordinatorLifecycleRecord extends ManagedOperationRecord { readonly abort: AbortController; readonly settle: (result: OperationResult) => void; readonly settled: Promise; + onTimeout?: () => void; cancellationRequested: boolean; + operationTimeout?: { cancel(): void }; reconciling?: Promise; timedOut: boolean; } @@ -30,6 +26,7 @@ export interface CoordinatorLifecycleOptions< Summary, > { readonly closeTimeoutMilliseconds?: number; + readonly operationTimeoutMilliseconds?: number; readonly publish?: (summary: Summary) => void; readonly summarize: (record: Record) => Summary; readonly terminalRetention?: number; @@ -63,10 +60,20 @@ class LifecycleAdapter< Record extends CoordinatorLifecycleRecord, Summary, > implements CoordinatorLifecycle { + readonly #operationTimeout: number | undefined; readonly #store: OperationLifecycleStore; + readonly #timeoutScheduler: TimeoutScheduler; #closePromise?: Promise; constructor(options: CoordinatorLifecycleOptions) { + this.#operationTimeout = + options.operationTimeoutMilliseconds === undefined + ? undefined + : validateTimeoutMilliseconds( + options.operationTimeoutMilliseconds, + 'operationTimeoutMilliseconds', + ); + this.#timeoutScheduler = options.timeoutScheduler ?? systemTimeoutScheduler; this.#store = createOperationLifecycleStore({ closeTimeoutMilliseconds: options.closeTimeoutMilliseconds, publish: options.publish, @@ -86,6 +93,7 @@ class LifecycleAdapter< this.#store.publish(record); if (!this.open) return 'interrupted' as const; record.phase = 'running'; + this.#armOperationTimeout(record); this.#store.publish(record); return this.open ? ('running' as const) : ('interrupted' as const); } @@ -180,6 +188,8 @@ class LifecycleAdapter< settle(record: Record, result: OperationResult) { this.#assertOwned(record); + record.operationTimeout?.cancel(); + record.operationTimeout = undefined; record.phase = 'terminal'; record.result = result; record.lease = result.kind === 'unknown_outcome'; @@ -197,6 +207,27 @@ class LifecycleAdapter< this.#store.publish(record); } + #armOperationTimeout(record: Record) { + if (this.#operationTimeout === undefined) return; + let fired = false; + const timeout = this.#timeoutScheduler.schedule( + this.#operationTimeout, + () => { + fired = true; + record.operationTimeout = undefined; + if (record.result !== undefined) return; + record.timedOut = true; + this.#requestCancellation(record); + record.onTimeout?.(); + }, + ); + record.operationTimeout = timeout; + if (fired || record.result !== undefined) { + timeout.cancel(); + record.operationTimeout = undefined; + } + } + #assertOwned(record: Record) { if (this.#store.get(record.id) !== record) { throw new Error('Operation record is not registered by this adapter.'); diff --git a/packages/repository-engine/src/operation-coordinator.test.ts b/packages/repository-engine/src/operation-coordinator.test.ts index b177ef6..2dbac35 100644 --- a/packages/repository-engine/src/operation-coordinator.test.ts +++ b/packages/repository-engine/src/operation-coordinator.test.ts @@ -8,11 +8,13 @@ import { } from '@codex-git/protocol'; import { - createOperationCoordinator, type CoordinatedOperation, - type OperationAdmission, type ReconciledOperationResult, } from './operation-coordinator.js'; +import { + createOperationSession, + type OperationSessionAdmission, +} from './operation-session.js'; const generation = (digit: string) => worktreeGenerationSchema.parse(`generation_${digit.repeat(32)}`); @@ -28,7 +30,7 @@ describe('operation coordinator admission', () => { const firstExecution = deferred(); const otherExecution = deferred(); const busyReconciliation = deferred(); - const coordinator = createOperationCoordinator(); + const coordinator = createOperationSession(); const firstGeneration = generation('1'); let blockedExecutions = 0; @@ -73,7 +75,7 @@ describe('operation coordinator admission', () => { }); it('rejects caller-selected coordination and incomplete kind targets at runtime', async () => { - const coordinator = createOperationCoordinator(); + const coordinator = createOperationSession(); const bypass = { ...commitOperation(generation('3'), 'refs/heads/topic', 'unused'), lane: { kind: 'remote' }, @@ -91,33 +93,11 @@ describe('operation coordinator admission', () => { ); }); - it('propagates Busy reconciliation failure without publishing Busy', async () => { - const activeExecution = deferred(); - const coordinator = createOperationCoordinator(); - const worktreeGeneration = generation('3'); - const active = await coordinator.dispatch( - stageOperation(worktreeGeneration, activeExecution.promise), - ); - const reconciliationFailure = new Error('Fresh state is unavailable.'); - - await expect( - coordinator.dispatch({ - ...commitOperation(worktreeGeneration, 'refs/heads/topic', 'blocked'), - reconcileBusy: async () => { - throw reconciliationFailure; - }, - }), - ).rejects.toBe(reconciliationFailure); - - activeExecution.resolve('active evidence'); - await coordinator.recover(acceptedId(active)); - }); - it('derives Repository lanes and mandatory cross-lane claims', async () => { const commitExecution = deferred(); const branchExecution = deferred(); const fetchExecution = deferred(); - const coordinator = createOperationCoordinator(); + const coordinator = createOperationSession(); const commitGeneration = generation('4'); const branchGeneration = generation('5'); @@ -191,7 +171,7 @@ describe('operation coordinator admission', () => { it('makes a Remote-tracking Branch conflict with Fetch for its exact Remote', async () => { const branchExecution = deferred(); - const coordinator = createOperationCoordinator(); + const coordinator = createOperationSession(); const remoteId = remote('2'); const branch = await coordinator.dispatch( branchOperation( @@ -219,7 +199,7 @@ describe('operation coordinator admission', () => { }); it('rejects Branch targets outside their declared ref namespace', async () => { - const coordinator = createOperationCoordinator(); + const coordinator = createOperationSession(); const localMismatch = branchOperation( generation('7'), localTarget('refs/remotes/origin/topic'), @@ -245,7 +225,7 @@ describe('operation coordinator admission', () => { it('reconciles an Unknown lease in the background without queueing admissions', async () => { const recovered = deferred(); - const coordinator = createOperationCoordinator(); + const coordinator = createOperationSession(); const worktreeGeneration = generation('8'); let reconciliations = 0; const first = await coordinator.dispatch({ @@ -301,7 +281,7 @@ describe('operation coordinator admission', () => { }); it('derives the terminal result from reconciliation rather than process state', async () => { - const coordinator = createOperationCoordinator(); + const coordinator = createOperationSession(); const admission = await coordinator.dispatch({ ...stageOperation(generation('9'), 'exit zero'), reconcile: async () => ({ @@ -394,7 +374,7 @@ function localTarget(fullName: string) { return { kind: 'local', fullName } as const; } -function acceptedId(admission: OperationAdmission) { +function acceptedId(admission: OperationSessionAdmission) { if (admission.kind !== 'accepted') throw new Error('Expected accepted'); return admission.operation.operationId; } diff --git a/packages/repository-engine/src/operation-coordinator.ts b/packages/repository-engine/src/operation-coordinator.ts index ba6d21b..7c66579 100644 --- a/packages/repository-engine/src/operation-coordinator.ts +++ b/packages/repository-engine/src/operation-coordinator.ts @@ -1,9 +1,6 @@ import { - createOpaqueIdAuthority, - operationResultSchema, remoteIdSchema, worktreeGenerationSchema, - type OperationId, type OperationResult, type RemoteId, type RepositorySnapshot, @@ -11,7 +8,6 @@ import { } from '@codex-git/protocol'; type OperationSummary = RepositorySnapshot['operations'][number]; -type RejectedResult = Extract; type WithoutId = Result extends OperationResult ? Omit : never; @@ -21,7 +17,8 @@ export type ReconciledOperationResult = WithoutId; export type OperationExecution = | { readonly kind: 'returned'; readonly evidence: Evidence } - | { readonly kind: 'threw'; readonly error: unknown }; + | { readonly kind: 'threw'; readonly error: unknown } + | { readonly kind: 'timed_out' }; export interface CoordinatedOperationSummary extends OperationSummary { readonly retryAllowed: boolean; @@ -85,190 +82,12 @@ type Target = export type CoordinatedOperation = Target & Hooks; -export type OperationAdmission = - | { - readonly kind: 'accepted'; - readonly operation: CoordinatedOperationSummary; - } - | { - readonly kind: 'rejected'; - readonly result: RejectedResult; - readonly conflicts: readonly CoordinatedOperationSummary[]; - }; - -export interface OperationCoordinator { - dispatch( - operation: CoordinatedOperation, - ): Promise; - recover(operationId: OperationId): Promise; -} - export interface OperationCoordination { readonly category: OperationSummary['category']; readonly claims: ReadonlySet; readonly lane: string; } -interface RecordState extends OperationCoordination { - readonly execute: (context: { signal: AbortSignal }) => Promise; - readonly id: OperationId; - readonly reconcile: ( - context: OperationReconciliationContext, - ) => Promise; - readonly settled: Promise; - readonly settle: (result: OperationResult) => void; - execution?: OperationExecution; - lease: boolean; - phase: OperationSummary['phase']; - reconciling?: Promise; - result?: OperationResult; -} - -class Coordinator implements OperationCoordinator { - readonly #ids = createOpaqueIdAuthority(); - readonly #operations = new Map(); - - async dispatch( - operation: CoordinatedOperation, - ): Promise { - const coordination = coordinateOperation(operation); - const conflicts = this.#conflicts(coordination); - - const unknown = conflicts.filter( - (record) => record.result?.kind === 'unknown_outcome', - ); - for (const record of unknown) void this.#reconcile(record); - - if (conflicts.length > 0) { - return this.#rejectBusy(operation, coordination, conflicts); - } - return this.#admit(operation, coordination); - } - - recover(operationId: OperationId): Promise { - const record = this.#operations.get(operationId); - if (record === undefined) { - throw new Error( - 'Operation result is not retained by this Repository Session.', - ); - } - if (record.result?.kind === 'unknown_outcome') { - return this.#reconcile(record); - } - return record.result === undefined - ? record.settled - : Promise.resolve(record.result); - } - - #admit( - operation: CoordinatedOperation, - coordination: OperationCoordination, - ): OperationAdmission { - const record = this.#record(operation, coordination, true, 'running'); - this.#operations.set(record.id, record); - void this.#run(record); - return { kind: 'accepted', operation: summary(record) }; - } - - async #rejectBusy( - operation: CoordinatedOperation, - coordination: OperationCoordination, - conflicts: readonly RecordState[], - ): Promise { - const conflictSummaries = conflicts.map(summary); - await operation.reconcileBusy({ conflicts: conflictSummaries }); - - const record = this.#record(operation, coordination, false, 'terminal'); - const result = withId(record.id, { - kind: 'rejected', - code: 'busy', - message: 'A conflicting operation is active.', - }) as RejectedResult; - record.result = result; - record.settle(result); - this.#operations.set(record.id, record); - return { kind: 'rejected', result, conflicts: conflictSummaries }; - } - - async #run(record: RecordState) { - try { - record.execution = { - kind: 'returned', - evidence: await record.execute({ - signal: new AbortController().signal, - }), - }; - } catch (error) { - record.execution = { kind: 'threw', error }; - } - await this.#reconcile(record); - } - - #reconcile(record: RecordState): Promise { - if (record.reconciling !== undefined) return record.reconciling; - if (record.execution === undefined) return record.settled; - - record.phase = 'reconciling'; - const promise = Promise.resolve() - .then(() => - record.reconcile({ - cancellationRequested: false, - execution: record.execution as OperationExecution, - timedOut: false, - }), - ) - .then((outcome) => this.#settle(record, outcome)) - .catch(() => this.#settle(record, unknownOutcome())); - record.reconciling = promise; - void promise.then(() => { - if (record.reconciling === promise) record.reconciling = undefined; - }); - return promise; - } - - #settle(record: RecordState, outcome: ReconciledOperationResult) { - const result = withId(record.id, outcome); - record.phase = 'terminal'; - record.result = result; - record.lease = result.kind === 'unknown_outcome'; - record.settle(result); - return result; - } - - #record( - operation: CoordinatedOperation, - coordination: OperationCoordination, - lease: boolean, - phase: OperationSummary['phase'], - ): RecordState { - const done = deferred(); - return { - ...coordination, - execute: (context) => operation.execute(context), - id: this.#ids.issue('operation'), - lease, - phase, - reconcile: (context) => - operation.reconcile( - context as OperationReconciliationContext, - ), - settle: done.resolve, - settled: done.promise, - }; - } - - #conflicts(candidate: OperationCoordination) { - return [...this.#operations.values()].filter( - (record) => - record.lease && operationCoordinationConflicts(record, candidate), - ); - } -} - -export function createOperationCoordinator(): OperationCoordinator { - return new Coordinator(); -} - export function coordinateOperation( operation: CoordinatedOperation, ): OperationCoordination { @@ -454,40 +273,6 @@ function overlaps(left: ReadonlySet, right: ReadonlySet) { return false; } -function summary(record: RecordState): CoordinatedOperationSummary { - return { - operationId: record.id, - category: record.category, - phase: record.phase, - progress: null, - retryAllowed: - record.phase === 'terminal' && - record.result !== undefined && - record.result.kind !== 'unknown_outcome', - }; -} - -function withId(id: OperationId, outcome: ReconciledOperationResult) { - return operationResultSchema.parse({ ...outcome, operationId: id }); -} - -function unknownOutcome(): ReconciledOperationResult { - return { - kind: 'unknown_outcome', - code: 'reconciliation_incomplete', - message: 'Reconciliation could not establish the operation outcome.', - recoveryAvailable: true, - }; -} - function key(...parts: readonly string[]) { return JSON.stringify(parts); } - -function deferred() { - let resolve!: (value: Value) => void; - const promise = new Promise((resolvePromise) => { - resolve = resolvePromise; - }); - return { promise, resolve }; -} diff --git a/packages/repository-engine/src/operation-lifecycle.ts b/packages/repository-engine/src/operation-lifecycle.ts index 4d48acb..131a8a1 100644 --- a/packages/repository-engine/src/operation-lifecycle.ts +++ b/packages/repository-engine/src/operation-lifecycle.ts @@ -6,7 +6,7 @@ import type { type OperationPhase = RepositorySnapshot['operations'][number]['phase']; -interface TimeoutScheduler { +export interface TimeoutScheduler { schedule( milliseconds: number, onTimeout: () => void, @@ -81,7 +81,7 @@ class LifecycleStore< #state: 'open' | 'closing' | 'closed' = 'open'; constructor(options: OperationLifecycleStoreOptions) { - this.#closeTimeout = milliseconds( + this.#closeTimeout = validateTimeoutMilliseconds( options.closeTimeoutMilliseconds ?? 250, 'closeTimeoutMilliseconds', ); @@ -267,7 +267,7 @@ function nonnegative(value: number, name: string) { return value; } -function milliseconds(value: number, name: string) { +export function validateTimeoutMilliseconds(value: number, name: string) { const validated = nonnegative(value, name); if (validated > 2_147_483_647) { throw new Error(`${name} exceeds the supported timer range.`); @@ -302,7 +302,7 @@ function waitBounded( }); } -const systemTimeoutScheduler: TimeoutScheduler = { +export const systemTimeoutScheduler: TimeoutScheduler = { schedule(milliseconds, onTimeout) { const timer = setTimeout(onTimeout, milliseconds); return { cancel: () => clearTimeout(timer) }; diff --git a/packages/repository-engine/src/operation-session.test.ts b/packages/repository-engine/src/operation-session.test.ts index 266b3cf..e99492a 100644 --- a/packages/repository-engine/src/operation-session.test.ts +++ b/packages/repository-engine/src/operation-session.test.ts @@ -351,6 +351,79 @@ describe('operation session', () => { }); }); + it('times out each admitted operation and reconciles before enabling retry', async () => { + const timeout = controlledTimeout(); + const contexts: OperationReconciliationContext[] = []; + const published: OperationSessionSummary[] = []; + const session = createOperationSession({ + operationTimeoutMilliseconds: 75, + publish: (summary) => published.push(summary), + timeoutScheduler: timeout.scheduler, + }); + let executionSignal: AbortSignal | undefined; + const lateExecution = deferred(); + const admission = await session.dispatch({ + ...stageOperation(generation('a'), 'unused'), + execute: ({ signal }) => { + executionSignal = signal; + return lateExecution.promise; + }, + reconcile: async (context) => { + contexts.push(context); + return { + kind: 'failed_known', + code: 'process_failed', + message: 'Timeout was reconciled from fresh state.', + }; + }, + }); + const operationId = acceptedId(admission); + + expect(timeout.delay).toBe(75); + timeout.trigger(); + expect(executionSignal?.aborted).toBe(true); + expect(published.at(-1)).toMatchObject({ + cancellationRequested: true, + phase: 'reconciling', + retryAllowed: false, + timedOut: true, + }); + + await expect(session.recover(operationId)).resolves.toMatchObject({ + kind: 'unknown_outcome', + code: 'reconciliation_incomplete', + }); + expect(contexts).toEqual([ + { + cancellationRequested: true, + execution: { kind: 'timed_out' }, + timedOut: true, + }, + ]); + expect(published.at(-1)).toMatchObject({ + cancellationRequested: true, + phase: 'terminal', + retryAllowed: false, + timedOut: true, + }); + const publicationsAtTimeout = published.length; + lateExecution.resolve('late evidence'); + await until(() => published.length > publicationsAtTimeout); + await expect(session.recover(operationId)).resolves.toMatchObject({ + kind: 'failed_known', + code: 'process_failed', + }); + expect(contexts.at(-1)).toEqual({ + cancellationRequested: true, + execution: { kind: 'returned', evidence: 'late evidence' }, + timedOut: true, + }); + expect(published.at(-1)).toMatchObject({ + phase: 'terminal', + retryAllowed: true, + }); + }); + it('drains on close despite publication failure and reentrancy', async () => { let reentrantClose: | ReturnType['close']> @@ -557,12 +630,17 @@ function deferred() { function controlledTimeout() { let callback: (() => void) | undefined; let cancellations = 0; + let delay: number | undefined; return { get cancellations() { return cancellations; }, + get delay() { + return delay; + }, scheduler: { - schedule(_milliseconds: number, onTimeout: () => void) { + schedule(milliseconds: number, onTimeout: () => void) { + delay = milliseconds; callback = onTimeout; return { cancel() { diff --git a/packages/repository-engine/src/operation-session.ts b/packages/repository-engine/src/operation-session.ts index 7b7a6f7..aeed672 100644 --- a/packages/repository-engine/src/operation-session.ts +++ b/packages/repository-engine/src/operation-session.ts @@ -163,17 +163,34 @@ class Session implements OperationSession { async #run(record: SessionRecord) { try { - record.execution = { - kind: 'returned', - evidence: await record.execute({ signal: record.abort.signal }), - }; + const evidence = await record.execute({ signal: record.abort.signal }); + if ( + record.execution === undefined || + record.execution.kind === 'timed_out' + ) { + record.execution = { kind: 'returned', evidence }; + } } catch (error) { - record.execution = { kind: 'threw', error }; + if ( + record.execution === undefined || + record.execution.kind === 'timed_out' + ) { + record.execution = { kind: 'threw', error }; + } } await this.#reconcile(record); + if (record.result?.kind === 'unknown_outcome') { + await this.#reconcile(record); + } } #reconcile(record: SessionRecord) { + if ( + record.result !== undefined && + record.result.kind !== 'unknown_outcome' + ) { + return Promise.resolve(record.result); + } if (record.execution === undefined && record.busyConflicts === undefined) { return record.settled; } @@ -189,11 +206,20 @@ class Session implements OperationSession { message: 'A conflicting operation is active.', }); } + const execution = record.execution as OperationExecution; const outcome = await record.reconcile({ cancellationRequested: record.cancellationRequested, - execution: record.execution as OperationExecution, + execution, timedOut: record.timedOut, }); + if (execution.kind === 'timed_out') { + return withId(record.id, { + kind: 'unknown_outcome', + code: 'reconciliation_incomplete', + message: 'The timed-out process has not confirmed termination.', + recoveryAvailable: true, + }); + } return withId(record.id, outcome); } @@ -203,7 +229,7 @@ class Session implements OperationSession { lease: boolean, ): SessionRecord { const done = deferred(); - return { + const record: SessionRecord = { ...coordination, abort: new AbortController(), cancellationRequested: false, @@ -220,6 +246,12 @@ class Session implements OperationSession { settled: done.promise, timedOut: false, }; + record.onTimeout = () => { + if (record.execution !== undefined) return; + record.execution = { kind: 'timed_out' }; + void this.#reconcile(record); + }; + return record; } #conflicts(candidate: OperationCoordination) { diff --git a/packages/repository-engine/src/repository-engine.ts b/packages/repository-engine/src/repository-engine.ts index 3fde7db..d03a638 100644 --- a/packages/repository-engine/src/repository-engine.ts +++ b/packages/repository-engine/src/repository-engine.ts @@ -17,11 +17,18 @@ import { type WorktreePorcelainRecord, } from './worktree-porcelain.js'; import { createGitEnvironment } from './git-environment.js'; +import { GitReadPolicy } from './git-read-policy.js'; import { createRepositoryObserver } from './repository-observation.js'; import { createRepositoryPublicationSession, - type RepositorySession, + type RepositoryRefreshScope, + type ScopedRepositoryPublicationSession, } from './repository-publication.js'; +import { createRefreshingRepositorySession } from './repository-refresh.js'; +import { + createRepositorySession, + type RepositorySession, +} from './repository-session.js'; import { cloneRemoteIdentityState, createRemoteIdentityState, @@ -118,17 +125,26 @@ export function createRepositoryEngine(): RepositoryEngine { const state: SessionState = { generation: 0, status: 'open' }; if (resolved === null) { - return { + const publication: ScopedRepositoryPublicationSession = { async snapshot() { beginSnapshot(state); return { kind: 'not_repository' }; }, async *subscribe() {}, + async requestRefresh() { + beginSnapshot(state); + return { kind: 'not_repository' } as const; + }, + async requestScopedRefresh() { + beginSnapshot(state); + return { kind: 'not_repository' } as const; + }, async close() { closeSession(state); ids.revokeAll(); }, }; + return createRepositorySession(publication); } const identity: RepositoryIdentityState = { @@ -137,33 +153,61 @@ export function createRepositoryEngine(): RepositoryEngine { generations: new Map(), remoteIdentity: createRemoteIdentityState(), }; - return createRepositoryPublicationSession({ - async read() { + const reads = new GitReadPolicy(4); + let observedDiscovery: RepositoryDiscovery | undefined; + const publication = createRepositoryPublicationSession({ + async read(signal, refreshGeneration, requestedScope) { const sessionGeneration = beginSnapshot(state); const candidateIdentity: RepositoryIdentityState = { ...identity, generations: new Map(identity.generations), remoteIdentity: cloneRemoteIdentityState(identity.remoteIdentity), }; - const result = await discoverRepository( - resolved, - candidateIdentity, - ids, - state, - sessionGeneration, - ); + let canReuseTopology = false; + let discovery: RepositoryDiscovery; + if ( + requestedScope.kind === 'worktrees' && + observedDiscovery !== undefined + ) { + canReuseTopology = true; + discovery = observedDiscovery; + } else { + discovery = ( + await discoverRepository( + resolved, + candidateIdentity, + ids, + state, + sessionGeneration, + signal, + reads, + refreshGeneration, + ) + ).repository; + } assertSessionGeneration(state, sessionGeneration); + const scope: RepositoryRefreshScope = canReuseTopology + ? requestedScope + : { kind: 'all' }; + const worktreeIds = + scope.kind === 'all' ? undefined : new Set(scope.worktreeIds); const observation = await createRepositoryObserver( runGit, ids, candidateIdentity.remoteIdentity, - ).observe(result.repository); + 4, + reads, + String(refreshGeneration), + ).observe(discovery, signal, worktreeIds); assertSessionGeneration(state, sessionGeneration); return { - discovery: result.repository, + discovery, observation, commit() { - identity.generations = candidateIdentity.generations; + if (!canReuseTopology) { + identity.generations = candidateIdentity.generations; + observedDiscovery = discovery; + } identity.remoteIdentity = candidateIdentity.remoteIdentity; }, }; @@ -179,6 +223,9 @@ export function createRepositoryEngine(): RepositoryEngine { ids.revokeAll(); }, }); + return createRefreshingRepositorySession( + createRepositorySession(publication), + ); }, }; } @@ -189,6 +236,9 @@ async function discoverRepository( ids: OpaqueIdAuthority, state: SessionState, sessionGeneration: number, + signal: AbortSignal, + reads: GitReadPolicy, + refreshGeneration: number, ): Promise<{ readonly kind: 'repository'; readonly repository: RepositoryDiscovery; @@ -200,7 +250,8 @@ async function discoverRepository( state, sessionGeneration, ); - const inventory = await runGit( + const inventory = await runDiscoveryRead( + reads, [ '--git-dir', resolved.commonGitDirectory, @@ -210,12 +261,21 @@ async function discoverRepository( '-z', ], true, + undefined, + signal, + refreshGeneration, ); assertSessionGeneration(state, sessionGeneration); const records = parseWorktreeListPorcelain(inventory); const resolvedRegistrations = await Promise.all( records.map((record) => - canonicalizeRegistration(record, resolved.commonGitDirectory), + canonicalizeRegistration( + record, + resolved.commonGitDirectory, + signal, + reads, + refreshGeneration, + ), ), ); assertSessionGeneration(state, sessionGeneration); @@ -347,6 +407,9 @@ async function resolveAnchor( async function canonicalizeRegistration( record: WorktreePorcelainRecord, commonGitDirectory: AbsolutePath, + signal: AbortSignal, + reads: GitReadPolicy, + refreshGeneration: number, ): Promise { let canonicalPathBytes: Uint8Array; let adminIdentity: string | null = null; @@ -382,7 +445,8 @@ async function canonicalizeRegistration( ); const gitDirectory = await realpath( decodeLine( - await runGit( + await runDiscoveryRead( + reads, [ '-C', canonicalPath, @@ -391,12 +455,16 @@ async function canonicalizeRegistration( '--git-dir', ], false, + undefined, + signal, + refreshGeneration, ), ), ); const resolvedCommonGitDirectory = await realpath( decodeLine( - await runGit( + await runDiscoveryRead( + reads, [ '-C', canonicalPath, @@ -405,6 +473,9 @@ async function canonicalizeRegistration( '--git-common-dir', ], false, + undefined, + signal, + refreshGeneration, ), ), ); @@ -412,7 +483,13 @@ async function canonicalizeRegistration( throw new WorktreeRegistrationMismatchError(); } if (gitDirectory !== commonGitDirectory) { - await assertWorktreeAdminBacklink(canonicalPathBytes, gitDirectory); + await assertWorktreeAdminBacklink( + canonicalPathBytes, + gitDirectory, + signal, + reads, + refreshGeneration, + ); } const gitDirectoryEvidence = fileIdentity(await stat(gitDirectory)); adminIdentity = `${gitDirectory}\0${gitDirectoryEvidence}`; @@ -445,9 +522,13 @@ async function canonicalizeRegistration( async function assertWorktreeAdminBacklink( canonicalPathBytes: Uint8Array, gitDirectory: string, + signal: AbortSignal, + reads: GitReadPolicy, + refreshGeneration: number, ): Promise { const adminControlPath = decodeLine( - await runGit( + await runDiscoveryRead( + reads, [ '--git-dir', gitDirectory, @@ -457,6 +538,9 @@ async function assertWorktreeAdminBacklink( 'gitdir', ], false, + undefined, + signal, + refreshGeneration, ), ); const backlinkTarget = stripLineEnding(await readFile(adminControlPath)); @@ -621,6 +705,7 @@ function runGit( args: readonly string[], allowLargeOutput: boolean, acceptedEmptyExitCode?: 1, + signal?: AbortSignal, ): Promise { return new Promise((resolvePromise, reject) => { execFile( @@ -630,6 +715,7 @@ function runGit( encoding: 'buffer', env: createGitEnvironment(), maxBuffer: allowLargeOutput ? GIT_OUTPUT_LIMIT_BYTES : 64 * 1_024, + signal, timeout: GIT_TIMEOUT_MILLISECONDS, windowsHide: true, }, @@ -655,6 +741,24 @@ function runGit( }); } +function runDiscoveryRead( + policy: GitReadPolicy, + args: readonly string[], + allowLargeOutput: boolean, + acceptedEmptyExitCode?: 1, + signal?: AbortSignal, + refreshGeneration = 0, +): Promise { + return policy.run( + `${refreshGeneration}:${JSON.stringify([ + allowLargeOutput, + acceptedEmptyExitCode, + args, + ])}`, + () => runGit(args, allowLargeOutput, acceptedEmptyExitCode, signal), + ); +} + function beginSnapshot(state: SessionState): number { if (state.status === 'invalid') { throw new Error( diff --git a/packages/repository-engine/src/repository-observation.ts b/packages/repository-engine/src/repository-observation.ts index e93832f..13da6d3 100644 --- a/packages/repository-engine/src/repository-observation.ts +++ b/packages/repository-engine/src/repository-observation.ts @@ -1,7 +1,11 @@ import { createHash } from 'node:crypto'; import { access } from 'node:fs/promises'; -import type { OpaqueIdAuthority } from '@codex-git/protocol'; +import type { + OpaqueIdAuthority, + RemoteId, + WorktreeId, +} from '@codex-git/protocol'; import { GitReadPolicy, runSelectedFirst } from './git-read-policy.js'; import type { @@ -45,8 +49,33 @@ export interface WorktreeStatusSummary { readonly untracked: number; } +export type UpstreamSnapshot = + | { + readonly kind: 'tracking'; + readonly remoteId: RemoteId; + readonly displayName: string; + readonly ref: { + readonly kind: 'remote_tracking'; + readonly fullName: string; + readonly objectId: string | null; + }; + readonly aheadBehind: + | { + readonly kind: 'cached'; + readonly ahead: number; + readonly behind: number; + } + | { readonly kind: 'unavailable' }; + } + | { readonly kind: 'unpublished' } + | { + readonly kind: 'not_applicable'; + readonly reason: 'detached_head' | 'unsupported_upstream'; + } + | { readonly kind: 'unavailable' }; + export interface WorktreeObservationError { - readonly code: 'git_read_failed' | 'git_output_too_large'; + readonly code: 'git_read_failed' | 'git_output_too_large' | 'not_observed'; readonly message: string; } @@ -57,6 +86,7 @@ export type WorktreeObservation = readonly head: DiscoveredHead; readonly index: IndexSnapshot; readonly status: WorktreeStatusSummary; + readonly upstream: UpstreamSnapshot; } | { readonly kind: 'unavailable'; @@ -71,16 +101,22 @@ export type WorktreeObservation = export interface RepositoryObservation { readonly shared: SharedRepositoryObservation; readonly worktrees: readonly WorktreeObservation[]; + readonly complete?: boolean; } export type GitReader = ( args: readonly string[], allowLargeOutput: boolean, acceptedEmptyExitCode?: 1, + signal?: AbortSignal, ) => Promise; export interface RepositoryObserver { - observe(discovery: RepositoryDiscovery): Promise; + observe( + discovery: RepositoryDiscovery, + signal?: AbortSignal, + worktreeIds?: ReadonlySet, + ): Promise; } export function createRepositoryObserver( @@ -88,23 +124,43 @@ export function createRepositoryObserver( ids: OpaqueIdAuthority, remoteIdentity: RemoteIdentityState, maximumConcurrency = DEFAULT_GIT_READ_CONCURRENCY, + readPolicy?: GitReadPolicy, + readNamespace = '', ): RepositoryObserver { - const reads = new GitReadPolicy(maximumConcurrency); + const reads = readPolicy ?? new GitReadPolicy(maximumConcurrency); + let observationGeneration = 0; return { - async observe(discovery) { + async observe(discovery, signal, worktreeIds) { + const readKeyPrefix = `${readNamespace}:${(observationGeneration += 1)}:`; let sharedBefore = await observeShared( discovery, reads, readGit, ids, remoteIdentity, + readKeyPrefix, + signal, ); for (let attempt = 0; attempt < MAX_COHERENCE_ATTEMPTS; attempt += 1) { + const selectedWorktrees = + worktreeIds === undefined + ? discovery.worktrees + : discovery.worktrees.filter(({ worktreeId }) => + worktreeIds.has(worktreeId), + ); const worktrees = await runSelectedFirst( - discovery.worktrees, + selectedWorktrees, ({ worktreeId }) => worktreeId === discovery.selectedWorktreeId, - (worktree) => observeWorktree(worktree, reads, readGit), + (worktree) => + observeWorktree( + worktree, + sharedBefore, + reads, + readGit, + readKeyPrefix, + signal, + ), ); const sharedAfter = await observeShared( discovery, @@ -112,9 +168,15 @@ export function createRepositoryObserver( readGit, ids, remoteIdentity, + readKeyPrefix, + signal, ); if (sameSharedObservation(sharedBefore, sharedAfter)) { - return { shared: sharedAfter, worktrees }; + return { + shared: sharedAfter, + worktrees, + complete: worktreeIds === undefined, + }; } sharedBefore = sharedAfter; } @@ -131,6 +193,8 @@ async function observeShared( readGit: GitReader, ids: OpaqueIdAuthority, remoteIdentity: RemoteIdentityState, + readKeyPrefix: string, + signal?: AbortSignal, ): Promise { const contextArgs = remoteContext(discovery); const [refsOutput, remoteObservation] = await Promise.all([ @@ -146,11 +210,22 @@ async function observeShared( 'refs/remotes', ], true, + undefined, + readKeyPrefix, + signal, ), observeRemotes( contextArgs, (args, allowLargeOutput, acceptedEmptyExitCode) => - runRead(reads, readGit, args, allowLargeOutput, acceptedEmptyExitCode), + runRead( + reads, + readGit, + args, + allowLargeOutput, + acceptedEmptyExitCode, + readKeyPrefix, + signal, + ), remoteIdentity, ids, ), @@ -186,8 +261,11 @@ function remoteContext(discovery: RepositoryDiscovery): readonly string[] { async function observeWorktree( worktree: DiscoveredWorktree, + shared: SharedRepositoryObservation, reads: GitReadPolicy, readGit: GitReader, + readKeyPrefix: string, + signal?: AbortSignal, ): Promise { if ( worktree.availability.kind === 'unavailable' || @@ -201,6 +279,8 @@ async function observeWorktree( worktree.canonicalPath, reads, readGit, + readKeyPrefix, + signal, ); const indexPathOutput = await runRead( reads, @@ -214,6 +294,9 @@ async function observeWorktree( 'index', ], false, + undefined, + readKeyPrefix, + signal, ); const observed = summarizeStatus(statusOutput, worktree.head); const indexPath = decodeLine(indexPathOutput); @@ -228,6 +311,7 @@ async function observeWorktree( locked: await pathExists(`${indexPath}.lock`), }, status: observed.status, + upstream: resolveUpstream(observed.upstream, shared), }; } catch (error) { return { @@ -242,6 +326,8 @@ async function readCoherentWorktree( canonicalPath: NonNullable, reads: GitReadPolicy, readGit: GitReader, + readKeyPrefix: string, + signal?: AbortSignal, ): Promise<{ readonly indexOutput: Uint8Array; readonly statusOutput: Uint8Array; @@ -252,6 +338,9 @@ async function readCoherentWorktree( readGit, ['-C', canonicalPath, 'ls-files', '--stage', '-z'], true, + undefined, + readKeyPrefix, + signal, ); const statusOutput = await runRead( reads, @@ -267,12 +356,18 @@ async function readCoherentWorktree( '--untracked-files=all', ], true, + undefined, + readKeyPrefix, + signal, ); const indexAfter = await runRead( reads, readGit, ['-C', canonicalPath, 'ls-files', '--stage', '-z'], true, + undefined, + readKeyPrefix, + signal, ); if (Buffer.from(indexBefore).equals(Buffer.from(indexAfter))) { return { indexOutput: indexAfter, statusOutput }; @@ -287,10 +382,16 @@ function runRead( args: readonly string[], allowLargeOutput: boolean, acceptedEmptyExitCode?: 1, + readKeyPrefix = '', + signal?: AbortSignal, ): Promise { return policy.run( - JSON.stringify([allowLargeOutput, acceptedEmptyExitCode, args]), - async () => readGit(args, allowLargeOutput, acceptedEmptyExitCode), + `${readKeyPrefix}${JSON.stringify([ + allowLargeOutput, + acceptedEmptyExitCode, + args, + ])}`, + async () => readGit(args, allowLargeOutput, acceptedEmptyExitCode, signal), ); } @@ -317,9 +418,16 @@ function parseRefs(output: Uint8Array): readonly RefSnapshot[] { function summarizeStatus( output: Uint8Array, fallbackHead: DiscoveredHead, -): { readonly head: DiscoveredHead; readonly status: WorktreeStatusSummary } { +): { + readonly head: DiscoveredHead; + readonly status: WorktreeStatusSummary; + readonly upstream: ObservedUpstream; +} { let branchHead: string | undefined; let branchObjectId: string | undefined; + let branchUpstream: string | undefined; + let branchAheadBehind: + { readonly ahead: number; readonly behind: number } | undefined; let conflicted = 0; let staged = 0; let unstaged = 0; @@ -336,6 +444,17 @@ function summarizeStatus( branchHead = header.slice('# branch.head '.length); } else if (header.startsWith('# branch.oid ')) { branchObjectId = header.slice('# branch.oid '.length); + } else if (header.startsWith('# branch.upstream ')) { + branchUpstream = header.slice('# branch.upstream '.length); + } else if (header.startsWith('# branch.ab ')) { + const match = /^# branch\.ab \+(\d+) -(\d+)$/u.exec(header); + if (match === null) { + throw new Error('Git status returned invalid Upstream divergence.'); + } + branchAheadBehind = { + ahead: Number(match[1]), + behind: Number(match[2]), + }; } continue; } @@ -367,6 +486,67 @@ function summarizeStatus( unstaged, untracked, }, + upstream: + branchHead === '(detached)' + ? { kind: 'detached' } + : branchUpstream === undefined + ? { kind: 'unpublished' } + : { + kind: 'configured', + displayName: branchUpstream, + aheadBehind: branchAheadBehind, + }, + }; +} + +type ObservedUpstream = + | { readonly kind: 'detached' } + | { readonly kind: 'unpublished' } + | { + readonly kind: 'configured'; + readonly displayName: string; + readonly aheadBehind?: { + readonly ahead: number; + readonly behind: number; + }; + }; + +function resolveUpstream( + observed: ObservedUpstream, + shared: SharedRepositoryObservation, +): UpstreamSnapshot { + if (observed.kind === 'detached') { + return { kind: 'not_applicable', reason: 'detached_head' }; + } + if (observed.kind === 'unpublished') { + return { kind: 'unpublished' }; + } + const remote = [...shared.remotes] + .sort((left, right) => right.displayName.length - left.displayName.length) + .find(({ displayName }) => + observed.displayName.startsWith(`${displayName}/`), + ); + if (remote === undefined) { + return { kind: 'not_applicable', reason: 'unsupported_upstream' }; + } + const fullName = `refs/remotes/${observed.displayName}`; + const ref = shared.refs.find( + (candidate) => + candidate.kind === 'remote_tracking' && candidate.fullName === fullName, + ); + return { + kind: 'tracking', + remoteId: remote.remoteId, + displayName: observed.displayName, + ref: { + kind: 'remote_tracking', + fullName, + objectId: ref?.objectId ?? null, + }, + aheadBehind: + observed.aheadBehind === undefined + ? { kind: 'unavailable' } + : { kind: 'cached', ...observed.aheadBehind }, }; } diff --git a/packages/repository-engine/src/repository-publication.ts b/packages/repository-engine/src/repository-publication.ts index 28aa138..fbbef7b 100644 --- a/packages/repository-engine/src/repository-publication.ts +++ b/packages/repository-engine/src/repository-publication.ts @@ -2,11 +2,14 @@ import type { RepositoryDiscovery } from './repository-engine.js'; import { InvalidationStream } from './invalidation-stream.js'; import { publishObservedFacts, + type PrivateRefsEvidence, type PublishedObservationWorktree, type PublishedRepositoryObservation, type WorktreeFreshness, } from './observation-publication.js'; import type { RepositoryObservation } from './repository-observation.js'; +import type { WorktreeId } from '@codex-git/protocol'; +import type { OperationSessionSummary } from './operation-session.js'; export interface RepositorySnapshot extends @@ -16,6 +19,7 @@ export interface RepositorySnapshot readonly topologyRevision: number; readonly refsRevision: number; readonly refresh: RefreshState; + readonly operations: readonly OperationSessionSummary[]; } export type PublishedWorktreeSnapshot = PublishedObservationWorktree; @@ -42,18 +46,37 @@ export type RepositoryOpenResult = readonly repository: RepositorySnapshot; }; -export interface RepositorySession { +export interface RepositoryPublicationSession { snapshot(): Promise; + requestRefresh(): Promise; subscribe(): AsyncIterable; close(): Promise; } -export interface RepositoryInvalidation { - readonly kind: 'repository'; - readonly repositoryRevision: number; - readonly refresh: RefreshState; +export type RepositoryRefreshScope = + | { readonly kind: 'all' } + | { + readonly kind: 'worktrees'; + readonly worktreeIds: readonly WorktreeId[]; + }; + +export interface ScopedRepositoryPublicationSession extends RepositoryPublicationSession { + requestScopedRefresh( + scope: RepositoryRefreshScope, + ): Promise; } +export type RepositoryInvalidation = + | { + readonly kind: 'repository'; + readonly repositoryRevision: number; + readonly refresh: RefreshState; + } + | { + readonly kind: 'operation'; + readonly operation: OperationSessionSummary; + }; + export class RepositorySessionFailure extends Error { constructor(readonly code: 'closed' | 'superseded') { super( @@ -72,69 +95,39 @@ interface PublicationCandidate { } interface PublicationSessionOptions { - read(): Promise; + read( + signal: AbortSignal, + refreshGeneration: number, + scope: RepositoryRefreshScope, + ): Promise; canRetainFailure(error: unknown): boolean; close(): void; } export function createRepositoryPublicationSession( options: PublicationSessionOptions, -): RepositorySession { +): ScopedRepositoryPublicationSession { const invalidations = new InvalidationStream(); let closed = false; let generation = 0; let published: RepositorySnapshot | undefined; - let privateRefsEvidence: string | undefined; + let privateRefsEvidence: PrivateRefsEvidence | undefined; + const activeReads = new Map(); - return { - async snapshot() { - if (closed) { - throw new RepositorySessionFailure('closed'); - } - const ownGeneration = ++generation; - let candidate: PublicationCandidate; - try { - candidate = await options.read(); - } catch (error) { - if (closed) { - throw new RepositorySessionFailure('closed'); - } - if (ownGeneration !== generation) { - if (published === undefined) { - throw new RepositorySessionFailure('superseded'); - } - return { kind: 'repository', repository: published }; - } - if (!options.canRetainFailure(error)) { - throw error; - } - const refreshError = classifyRefreshError(error); - if (published === undefined) { - return { - kind: 'failed', - refresh: deepFreeze({ kind: 'failed', error: refreshError }), - }; - } - const staleRefresh = deepFreeze({ - kind: 'stale', - error: refreshError, - } satisfies RefreshState); - if (sameExternalState(published.refresh, staleRefresh)) { - return { kind: 'repository', repository: published }; - } - const stale = deepFreeze({ - ...published, - repositoryRevision: published.repositoryRevision + 1, - refresh: staleRefresh, - } satisfies RepositorySnapshot); - published = stale; - invalidations.publish({ - kind: 'repository', - repositoryRevision: stale.repositoryRevision, - refresh: stale.refresh, - }); - return { kind: 'repository', repository: stale }; - } + const readSnapshot = async ( + scope: RepositoryRefreshScope, + ): Promise => { + if (closed) { + throw new RepositorySessionFailure('closed'); + } + const ownGeneration = ++generation; + const controller = new AbortController(); + activeReads.set(ownGeneration, controller); + let candidate: PublicationCandidate; + try { + candidate = await options.read(controller.signal, ownGeneration, scope); + } catch (error) { + activeReads.delete(ownGeneration); if (closed) { throw new RepositorySessionFailure('closed'); } @@ -144,25 +137,75 @@ export function createRepositoryPublicationSession( } return { kind: 'repository', repository: published }; } - const nextCandidate = publishCandidate( - candidate.discovery, - published, - candidate.observation, - privateRefsEvidence, - ); - const next = nextCandidate.snapshot; - candidate.commit(); - if (published !== undefined && sameExternalState(published, next)) { + if (!options.canRetainFailure(error)) { + throw error; + } + const refreshError = classifyRefreshError(error); + if (published === undefined) { + return { + kind: 'failed', + refresh: deepFreeze({ kind: 'failed', error: refreshError }), + }; + } + const staleRefresh = deepFreeze({ + kind: 'stale', + error: refreshError, + } satisfies RefreshState); + if (sameExternalState(published.refresh, staleRefresh)) { return { kind: 'repository', repository: published }; } - published = next; - privateRefsEvidence = nextCandidate.privateRefsEvidence; + const stale = deepFreeze({ + ...published, + repositoryRevision: published.repositoryRevision + 1, + refresh: staleRefresh, + } satisfies RepositorySnapshot); + published = stale; invalidations.publish({ kind: 'repository', - repositoryRevision: next.repositoryRevision, - refresh: next.refresh, + repositoryRevision: stale.repositoryRevision, + refresh: stale.refresh, }); - return { kind: 'repository', repository: next }; + return { kind: 'repository', repository: stale }; + } + activeReads.delete(ownGeneration); + if (closed) { + throw new RepositorySessionFailure('closed'); + } + if (ownGeneration !== generation) { + if (published === undefined) { + throw new RepositorySessionFailure('superseded'); + } + return { kind: 'repository', repository: published }; + } + const nextCandidate = publishCandidate( + candidate.discovery, + published, + candidate.observation, + privateRefsEvidence, + ); + const next = nextCandidate.snapshot; + candidate.commit(); + abortSupersededReads(activeReads, ownGeneration); + if (published !== undefined && sameExternalState(published, next)) { + return { kind: 'repository', repository: published }; + } + published = next; + privateRefsEvidence = nextCandidate.privateRefsEvidence; + invalidations.publish({ + kind: 'repository', + repositoryRevision: next.repositoryRevision, + refresh: next.refresh, + }); + return { kind: 'repository', repository: next }; + }; + + return { + snapshot: () => readSnapshot({ kind: 'all' }), + requestRefresh() { + return readSnapshot({ kind: 'all' }); + }, + requestScopedRefresh(scope) { + return readSnapshot(scope); }, subscribe() { return invalidations.subscribe(); @@ -173,12 +216,26 @@ export function createRepositoryPublicationSession( } closed = true; generation += 1; + for (const controller of activeReads.values()) controller.abort(); + activeReads.clear(); options.close(); invalidations.close(); }, }; } +function abortSupersededReads( + activeReads: Map, + publishedGeneration: number, +): void { + for (const [readGeneration, controller] of activeReads) { + if (readGeneration < publishedGeneration) { + controller.abort(); + activeReads.delete(readGeneration); + } + } +} + function classifyRefreshError(error: unknown): RefreshError { if ( error instanceof Error && @@ -208,10 +265,10 @@ function publishCandidate( discovery: RepositoryDiscovery, previous?: RepositorySnapshot, observation?: RepositoryObservation, - previousPrivateRefsEvidence?: string, + previousPrivateRefsEvidence?: PrivateRefsEvidence, ): { readonly snapshot: RepositorySnapshot; - readonly privateRefsEvidence: string; + readonly privateRefsEvidence: PrivateRefsEvidence; } { const { refsChanged, @@ -258,6 +315,7 @@ function publishCandidate( ? 1 : previous.refsRevision + (refsChanged ? 1 : 0), refresh: { kind: 'fresh' }, + operations: previous?.operations ?? [], ...publishedObservation, }), }; diff --git a/packages/repository-engine/src/repository-refresh.ts b/packages/repository-engine/src/repository-refresh.ts new file mode 100644 index 0000000..db5e2bf --- /dev/null +++ b/packages/repository-engine/src/repository-refresh.ts @@ -0,0 +1,214 @@ +import { watch, type FSWatcher } from 'node:fs'; + +import type { WorktreeId } from '@codex-git/protocol'; + +import { + type RepositoryOpenResult, + type RepositoryRefreshScope, +} from './repository-publication.js'; +import type { + InternalRepositorySession, + RepositorySession, +} from './repository-session.js'; + +const FILESYSTEM_DEBOUNCE_MILLISECONDS = 75; +const SELECTED_WORKTREE_POLL_MILLISECONDS = 1_000; +const NON_SELECTED_WORKTREE_POLL_MILLISECONDS = 5_000; +const DISCOVERY_FALLBACK_POLL_MILLISECONDS = 30_000; + +export function createRefreshingRepositorySession( + delegate: InternalRepositorySession, +): RepositorySession { + let closed = false; + let debounceTimer: NodeJS.Timeout | undefined; + let backgroundRefresh: Promise | undefined; + let pendingScope: RepositoryRefreshScope | undefined; + let requestedRefresh: Promise | undefined; + let watchedTopology = ''; + let selectedWorktreeId: WorktreeId | null = null; + let nonSelectedWorktreeIds: readonly WorktreeId[] = []; + let nonSelectedCursor = 0; + let foregroundRefreshes = 0; + const watchers: FSWatcher[] = []; + const pollTimers: NodeJS.Timeout[] = []; + + const observeResult = async ( + read: () => Promise, + ): Promise => { + const result = await read(); + if (!closed && result.kind === 'repository') configureObservation(result); + return result; + }; + + const startPendingBackgroundRefresh = () => { + if (foregroundRefreshes !== 0 || pendingScope === undefined) return; + const scope = pendingScope; + pendingScope = undefined; + startBackgroundRefresh(scope); + }; + + const startBackgroundRefresh = (scope: RepositoryRefreshScope) => { + if (closed) return; + if (backgroundRefresh !== undefined || foregroundRefreshes > 0) { + pendingScope = mergeScopes(pendingScope, scope); + return; + } + backgroundRefresh = observeResult(() => + delegate.requestScopedRefresh(scope), + ) + .then(() => undefined) + .catch(() => undefined) + .finally(() => { + backgroundRefresh = undefined; + startPendingBackgroundRefresh(); + }); + }; + + const scheduleFilesystemRefresh = (scope: RepositoryRefreshScope) => { + if (closed) return; + pendingScope = mergeScopes(pendingScope, scope); + if (debounceTimer !== undefined) clearTimeout(debounceTimer); + debounceTimer = setTimeout(() => { + debounceTimer = undefined; + const next = pendingScope ?? { kind: 'all' }; + pendingScope = undefined; + startBackgroundRefresh(next); + }, FILESYSTEM_DEBOUNCE_MILLISECONDS); + debounceTimer.unref(); + }; + + const configureObservation = ( + result: Extract, + ) => { + selectedWorktreeId = result.repository.selectedWorktreeId; + nonSelectedWorktreeIds = result.repository.worktrees + .filter(({ worktreeId }) => worktreeId !== selectedWorktreeId) + .map(({ worktreeId }) => worktreeId); + if (nonSelectedCursor >= nonSelectedWorktreeIds.length) { + nonSelectedCursor = 0; + } + + const paths: Array<{ + readonly path: string; + readonly scope: RepositoryRefreshScope; + }> = [ + { + path: result.repository.commonGitDirectory, + scope: { kind: 'all' }, + }, + ...result.repository.worktrees + .filter( + ({ availability, canonicalPath }) => + availability.kind === 'available' && canonicalPath !== null, + ) + .map(({ canonicalPath, worktreeId }) => ({ + path: canonicalPath as string, + scope: { kind: 'worktrees', worktreeIds: [worktreeId] } as const, + })), + ]; + const topology = JSON.stringify( + paths + .map(({ path, scope }) => ({ path, scope })) + .sort((a, b) => a.path.localeCompare(b.path)), + ); + if (topology !== watchedTopology) { + watchedTopology = topology; + for (const watcher of watchers.splice(0)) watcher.close(); + for (const { path, scope } of paths) { + try { + const watcher = watch( + path, + { + persistent: false, + recursive: process.platform === 'darwin', + }, + () => scheduleFilesystemRefresh(scope), + ); + watcher.on('error', () => watcher.close()); + watchers.push(watcher); + } catch { + // Polling remains the correctness fallback for an unsupported path. + } + } + } + if (pollTimers.length === 0) configurePolling(); + }; + + const configurePolling = () => { + const selected = setInterval(() => { + if (selectedWorktreeId !== null) { + startBackgroundRefresh({ + kind: 'worktrees', + worktreeIds: [selectedWorktreeId], + }); + } + }, SELECTED_WORKTREE_POLL_MILLISECONDS); + const nonSelected = setInterval(() => { + const worktreeId = nonSelectedWorktreeIds[nonSelectedCursor]; + if (worktreeId !== undefined) { + nonSelectedCursor = + (nonSelectedCursor + 1) % nonSelectedWorktreeIds.length; + startBackgroundRefresh({ + kind: 'worktrees', + worktreeIds: [worktreeId], + }); + } + }, NON_SELECTED_WORKTREE_POLL_MILLISECONDS); + const discovery = setInterval( + () => startBackgroundRefresh({ kind: 'all' }), + DISCOVERY_FALLBACK_POLL_MILLISECONDS, + ); + for (const timer of [selected, nonSelected, discovery]) { + timer.unref(); + pollTimers.push(timer); + } + }; + + const foreground = async ( + read: () => Promise, + ): Promise => { + foregroundRefreshes += 1; + try { + return await observeResult(read); + } finally { + foregroundRefreshes -= 1; + startPendingBackgroundRefresh(); + } + }; + + return { + snapshot: () => foreground(() => delegate.snapshot()), + requestRefresh() { + if (requestedRefresh !== undefined) return requestedRefresh; + requestedRefresh = foreground(() => delegate.requestRefresh()).finally( + () => { + requestedRefresh = undefined; + }, + ); + return requestedRefresh; + }, + subscribe: () => delegate.subscribe(), + cancelOperation: (operationId) => delegate.cancelOperation(operationId), + recoverOperation: (operationId) => delegate.recoverOperation(operationId), + async close() { + if (closed) return; + closed = true; + if (debounceTimer !== undefined) clearTimeout(debounceTimer); + for (const timer of pollTimers.splice(0)) clearInterval(timer); + for (const watcher of watchers.splice(0)) watcher.close(); + await delegate.close(); + }, + }; +} + +function mergeScopes( + left: RepositoryRefreshScope | undefined, + right: RepositoryRefreshScope, +): RepositoryRefreshScope { + if (left === undefined) return right; + if (left.kind === 'all' || right.kind === 'all') return { kind: 'all' }; + return { + kind: 'worktrees', + worktreeIds: [...new Set([...left.worktreeIds, ...right.worktreeIds])], + }; +} diff --git a/packages/repository-engine/src/repository-session.ts b/packages/repository-engine/src/repository-session.ts new file mode 100644 index 0000000..6288a61 --- /dev/null +++ b/packages/repository-engine/src/repository-session.ts @@ -0,0 +1,123 @@ +import type { OperationId, OperationResult } from '@codex-git/protocol'; + +import { InvalidationStream } from './invalidation-stream.js'; +import { + createOperationSession, + type OperationSessionSummary, +} from './operation-session.js'; +import type { + RepositoryInvalidation, + RepositoryOpenResult, + RepositoryPublicationSession, + RepositoryRefreshScope, + RepositorySnapshot, + ScopedRepositoryPublicationSession, +} from './repository-publication.js'; + +const OPERATION_TIMEOUT_MILLISECONDS = 30_000; + +export interface RepositorySession extends RepositoryPublicationSession { + cancelOperation(operationId: OperationId): Promise; + recoverOperation(operationId: OperationId): Promise; +} + +export interface InternalRepositorySession + extends RepositorySession, ScopedRepositoryPublicationSession {} + +export function createRepositorySession( + delegate: ScopedRepositoryPublicationSession, +): InternalRepositorySession { + const invalidations = new InvalidationStream(); + const operationSummaries = new Map(); + let closed = false; + let latestBase: RepositorySnapshot | undefined; + let latestBaseRevision = 0; + let latest: RepositorySnapshot | undefined; + let repositoryRevision = 0; + let operationEvidence = '[]'; + let postOperationRefresh: Promise | undefined; + + const publishCurrent = (base: RepositorySnapshot): RepositorySnapshot => { + const operations = [...operationSummaries.values()]; + const nextOperationEvidence = JSON.stringify(operations); + const changed = + latest === undefined || + base.repositoryRevision !== latestBaseRevision || + nextOperationEvidence !== operationEvidence; + if (!changed) return latest as RepositorySnapshot; + repositoryRevision = latest === undefined ? 1 : repositoryRevision + 1; + latestBaseRevision = base.repositoryRevision; + operationEvidence = nextOperationEvidence; + latest = deepFreeze({ + ...base, + repositoryRevision, + operations, + }); + invalidations.publish({ + kind: 'repository', + repositoryRevision, + refresh: latest.refresh, + }); + return latest; + }; + + const observe = async ( + request: () => Promise, + ): Promise => { + const result = await request(); + if (result.kind !== 'repository') return result; + latestBase = result.repository; + return { kind: 'repository', repository: publishCurrent(latestBase) }; + }; + + const schedulePostOperationRefresh = () => { + if (closed || postOperationRefresh !== undefined) return; + postOperationRefresh = observe(() => delegate.requestRefresh()) + .then(() => undefined) + .catch(() => undefined) + .finally(() => { + postOperationRefresh = undefined; + }); + }; + + const operations = createOperationSession({ + operationTimeoutMilliseconds: OPERATION_TIMEOUT_MILLISECONDS, + publish(summary) { + invalidations.publish({ kind: 'operation', operation: summary }); + operationSummaries.set(summary.operationId, summary); + if (latestBase !== undefined) publishCurrent(latestBase); + if (summary.phase === 'terminal') schedulePostOperationRefresh(); + }, + }); + + return { + snapshot: () => observe(() => delegate.snapshot()), + requestRefresh: () => observe(() => delegate.requestRefresh()), + requestScopedRefresh: (scope: RepositoryRefreshScope) => + observe(() => delegate.requestScopedRefresh(scope)), + subscribe: () => invalidations.subscribe(), + async cancelOperation(operationId) { + const result = await operations.cancel(operationId); + await observe(() => delegate.requestRefresh()); + return result; + }, + async recoverOperation(operationId) { + const result = await operations.recover(operationId); + await observe(() => delegate.requestRefresh()); + return result; + }, + async close() { + if (closed) return; + closed = true; + await operations.close(); + await delegate.close(); + invalidations.close(); + }, + }; +} + +function deepFreeze(value: T): T { + if (value === null || typeof value !== 'object') return value; + for (const child of Object.values(value)) deepFreeze(child); + return Object.freeze(value); +} diff --git a/tests/e2e/protocol-runtime.e2e.test.ts b/tests/e2e/protocol-runtime.e2e.test.ts index 4a75692..09b6409 100644 --- a/tests/e2e/protocol-runtime.e2e.test.ts +++ b/tests/e2e/protocol-runtime.e2e.test.ts @@ -1,3 +1,6 @@ +import { writeFile } from 'node:fs/promises'; +import { join } from 'node:path'; + import { afterEach, describe, expect, it } from 'vitest'; import { @@ -6,10 +9,19 @@ import { } from '@codex-git/launcher'; import { PROTOCOL_VERSION_HEADER } from '@codex-git/protocol'; +import { + createTemporaryGitRepository, + type TemporaryGitRepository, +} from '../fixtures/temporary-git-repository.js'; + const runtimes: StandaloneRuntime[] = []; +const repositories: TemporaryGitRepository[] = []; afterEach(async () => { await Promise.all(runtimes.splice(0).map((runtime) => runtime.close())); + await Promise.all( + repositories.splice(0).map((repository) => repository.dispose()), + ); }); describe('protocol runtime composition', () => { @@ -52,4 +64,55 @@ describe('protocol runtime composition', () => { tokenIsOpaque: true, }); }); + + it('streams Repository invalidations after an external selected Worktree change', async () => { + const repository = await createTemporaryGitRepository(); + repositories.push(repository); + await repository.git('config', 'user.name', 'Codex Git Tests'); + await repository.git('config', 'user.email', 'codex-git@example.test'); + await writeFile(join(repository.path, 'README.md'), 'fixture\n'); + await repository.git('add', '--', 'README.md'); + await repository.git('commit', '--quiet', '-m', 'Create fixture'); + const runtime = await startStandaloneRuntime({ + projectPath: repository.path, + surfacePort: 0, + }); + runtimes.push(runtime); + const eventsUrl = new URL( + runtime.sessionUrl.pathname.replace(/\/session$/u, '/events'), + runtime.sessionUrl, + ); + const response = await fetch(eventsUrl, { + headers: { origin: runtime.surfaceUrl.origin }, + }); + const reader = response.body?.getReader(); + if (reader === undefined) throw new Error('SSE response body is absent.'); + + await writeFile(join(repository.path, 'external.txt'), 'changed\n'); + const frame = await readFrameWithin(reader, 2_000); + + expect(frame).toContain('event: invalidation'); + expect(frame).toMatch(/"kind":"repository_revision"/u); + await reader.cancel(); + }); }); + +async function readFrameWithin( + reader: ReadableStreamDefaultReader, + milliseconds: number, +): Promise { + const decoder = new TextDecoder(); + let content = ''; + const deadline = new Promise((_, reject) => + setTimeout( + () => reject(new Error('Timed out waiting for an SSE invalidation.')), + milliseconds, + ), + ); + while (!content.includes('\n\n')) { + const next = await Promise.race([reader.read(), deadline]); + if (next.done) throw new Error('SSE stream closed before invalidation.'); + content += decoder.decode(next.value, { stream: true }); + } + return content; +} diff --git a/tests/integration/repository-discovery.integration.test.ts b/tests/integration/repository-discovery.integration.test.ts index 24a30a5..4c74183 100644 --- a/tests/integration/repository-discovery.integration.test.ts +++ b/tests/integration/repository-discovery.integration.test.ts @@ -122,11 +122,14 @@ describe('Repository Engine discovery', () => { try { const snapshot = session.snapshot(); + const rejectedSnapshot = expect(snapshot).rejects.toThrow( + 'Repository Session is closed.', + ); await waitForPath(marker); await session.close(); await writeFile(release, 'continue\n'); - await expect(snapshot).rejects.toThrow('Repository Session is closed.'); + await rejectedSnapshot; await expect(session.snapshot()).rejects.toThrow( 'Repository Session is closed.', ); diff --git a/tests/integration/repository-observation.integration.test.ts b/tests/integration/repository-observation.integration.test.ts index f66b422..f029423 100644 --- a/tests/integration/repository-observation.integration.test.ts +++ b/tests/integration/repository-observation.integration.test.ts @@ -207,6 +207,178 @@ describe('Repository observation', () => { await session.close(); }); + it('publishes an effective Upstream with opaque Remote and cached divergence evidence', async () => { + const repository = await createRepositoryWithCommit(); + const branch = ( + await repository.git('branch', '--show-current') + ).stdout.trim(); + await repository.git( + 'remote', + 'add', + 'origin', + 'https://user:secret@example.test/team/repository.git', + ); + await repository.git('branch', 'upstream-fixture'); + await repository.git('switch', '--quiet', 'upstream-fixture'); + await writeFile(join(repository.path, 'upstream.txt'), 'upstream\n'); + await repository.git('add', '--', 'upstream.txt'); + await repository.git('commit', '--quiet', '-m', 'Advance Upstream fixture'); + const upstreamObjectId = ( + await repository.git('rev-parse', 'HEAD') + ).stdout.trim(); + await repository.git('switch', '--quiet', branch); + await writeFile(join(repository.path, 'local.txt'), 'local\n'); + await repository.git('add', '--', 'local.txt'); + await repository.git('commit', '--quiet', '-m', 'Advance Local fixture'); + await repository.git( + 'update-ref', + `refs/remotes/origin/${branch}`, + upstreamObjectId, + ); + await repository.git('config', `branch.${branch}.remote`, 'origin'); + await repository.git( + 'config', + `branch.${branch}.merge`, + `refs/heads/${branch}`, + ); + const session = await createRepositoryEngine().open( + repository.path as AbsolutePath, + ); + + const snapshot = await snapshotRepository(session); + const remote = snapshot.remotes[0]!; + + expect(snapshot.worktrees[0]?.upstream).toEqual({ + kind: 'tracking', + remoteId: remote.remoteId, + displayName: `origin/${branch}`, + ref: { + kind: 'remote_tracking', + fullName: `refs/remotes/origin/${branch}`, + objectId: upstreamObjectId, + }, + aheadBehind: { kind: 'cached', ahead: 1, behind: 1 }, + }); + expect(JSON.stringify(snapshot)).not.toMatch(/secret|https:/u); + await session.close(); + }); + + it('resolves the effective Upstream of an attached unborn Branch', async () => { + const repository = await createTemporaryGitRepository(); + repositories.push(repository); + const branch = ( + await repository.git('symbolic-ref', '--short', 'HEAD') + ).stdout.trim(); + await repository.git( + 'remote', + 'add', + 'origin', + 'ssh://git@example.test/team/repository.git', + ); + await repository.git('config', `branch.${branch}.remote`, 'origin'); + await repository.git( + 'config', + `branch.${branch}.merge`, + `refs/heads/${branch}`, + ); + const session = await createRepositoryEngine().open( + repository.path as AbsolutePath, + ); + + const snapshot = await snapshotRepository(session); + const worktree = snapshot.worktrees[0]!; + + expect(worktree.head).toMatchObject({ + kind: 'local_branch', + displayName: branch, + objectId: null, + }); + expect(worktree.upstream).toEqual({ + kind: 'tracking', + remoteId: snapshot.remotes[0]!.remoteId, + displayName: `origin/${branch}`, + ref: { + kind: 'remote_tracking', + fullName: `refs/remotes/origin/${branch}`, + objectId: null, + }, + aheadBehind: { kind: 'unavailable' }, + }); + await session.close(); + }); + + it('does not treat an unrelated detached Worktree topology change as Upstream evidence', async () => { + const repository = await createRepositoryWithCommit(); + const linkedPath = `${repository.path}-detached-upstream-evidence`; + externalPaths.push(linkedPath); + const session = await createRepositoryEngine().open( + repository.path as AbsolutePath, + ); + const initial = await snapshotRepository(session); + const initialMain = initial.worktrees.find(({ role }) => role === 'main')!; + + await repository.git('worktree', 'add', '--quiet', '--detach', linkedPath); + const changed = await snapshotRepository(session); + const changedMain = changed.worktrees.find(({ role }) => role === 'main')!; + + expect(changed.repositoryRevision).toBe(initial.repositoryRevision + 1); + expect(changed.topologyRevision).toBe(initial.topologyRevision + 1); + expect(changed.refsRevision).toBe(initial.refsRevision); + expect(changedMain.worktreeRevision).toBe(initialMain.worktreeRevision); + expect( + changed.worktrees.find(({ role }) => role === 'linked')?.upstream, + ).toEqual({ kind: 'not_applicable', reason: 'detached_head' }); + await session.close(); + }); + + it('publishes external Upstream configuration changes only on shared revision axes', async () => { + const repository = await createRepositoryWithCommit(); + const branch = ( + await repository.git('branch', '--show-current') + ).stdout.trim(); + await repository.git( + 'remote', + 'add', + 'origin', + 'https://example.test/team/repository.git', + ); + const session = await createRepositoryEngine().open( + repository.path as AbsolutePath, + ); + const initial = await snapshotRepository(session); + const events = session.subscribe()[Symbol.asyncIterator](); + expect(initial.worktrees[0]?.upstream).toEqual({ kind: 'unpublished' }); + + await repository.git('config', `branch.${branch}.remote`, 'origin'); + await repository.git( + 'config', + `branch.${branch}.merge`, + `refs/heads/${branch}`, + ); + const changed = await snapshotRepository(session); + + expect(changed.worktrees[0]?.upstream).toMatchObject({ + kind: 'tracking', + remoteId: initial.remotes[0]!.remoteId, + displayName: `origin/${branch}`, + }); + expect(changed.repositoryRevision).toBe(initial.repositoryRevision + 1); + expect(changed.refsRevision).toBe(initial.refsRevision + 1); + expect(changed.topologyRevision).toBe(initial.topologyRevision); + expect(changed.worktrees[0]?.worktreeRevision).toBe( + initial.worktrees[0]?.worktreeRevision, + ); + await expect(events.next()).resolves.toEqual({ + done: false, + value: { + kind: 'repository', + repositoryRevision: changed.repositoryRevision, + refresh: { kind: 'fresh' }, + }, + }); + await session.close(); + }); + it('versions external Index, status, HEAD, and attached ref changes on their owning axes', async () => { const repository = await createRepositoryWithCommit(); const session = await createRepositoryEngine().open( diff --git a/tests/integration/repository-publication.integration.test.ts b/tests/integration/repository-publication.integration.test.ts index bcebfb0..0f9cfae 100644 --- a/tests/integration/repository-publication.integration.test.ts +++ b/tests/integration/repository-publication.integration.test.ts @@ -2,6 +2,7 @@ import { access, chmod, mkdtemp, + readFile, realpath, rm, writeFile, @@ -123,6 +124,35 @@ describe('Repository snapshot publication', () => { await session.close(); }); + it('publishes a selected Worktree change without a manual snapshot request', async () => { + const repository = await createRepositoryWithCommit(); + const session = await createRepositoryEngine().open( + asAbsolutePath(repository.path), + ); + const initial = await snapshotRepository(session); + const events = session.subscribe()[Symbol.asyncIterator](); + + await writeFile( + join(repository.path, 'automatic-refresh.txt'), + 'changed\n', + ); + + const event = await nextEventWithin(events, 2_000); + expect(event).toMatchObject({ + done: false, + value: { + kind: 'repository', + repositoryRevision: initial.repositoryRevision + 1, + }, + }); + const refreshed = await snapshotRepository(session); + expect(refreshed.worktrees[0]).toMatchObject({ + worktreeRevision: initial.worktrees[0]!.worktreeRevision + 1, + status: { untracked: 1 }, + }); + await session.close(); + }); + it('does not publish or notify a superseded snapshot that completes late', async () => { const repository = await createRepositoryWithCommit(); const session = await createRepositoryEngine().open( @@ -164,7 +194,6 @@ describe('Repository snapshot publication', () => { }, }); - await writeFile(release, 'continue\n'); const late = await older; expect(late).toEqual({ kind: 'repository', repository: newer }); await expect(noEvent(events)).resolves.toBe(true); @@ -182,6 +211,55 @@ describe('Repository snapshot publication', () => { } }); + it('deduplicates equivalent discovery reads across overlapping Refresh generations', async () => { + const repository = await createRepositoryWithCommit(); + const session = await createRepositoryEngine().open( + asAbsolutePath(repository.path), + ); + await snapshotRepository(session); + const directory = await mkdtemp(join(tmpdir(), 'codex-git-dedup-read-')); + externalPaths.push(directory); + const calls = join(directory, 'calls'); + const release = join(directory, 'release'); + const wrapper = join(directory, 'git'); + await writeFile( + wrapper, + [ + '#!/bin/sh', + 'if [ "$1" = "--git-dir" ] && [ "$3" = "worktree" ]; then', + ' printf "call\\n" >> "$CODEX_GIT_DEDUP_CALLS"', + ' while [ ! -e "$CODEX_GIT_DEDUP_RELEASE" ]; do sleep 0.01; done', + 'fi', + 'exec /usr/bin/git "$@"', + '', + ].join('\n'), + ); + await chmod(wrapper, 0o755); + const previousPath = process.env.PATH; + process.env.PATH = `${directory}:${previousPath ?? ''}`; + process.env.CODEX_GIT_DEDUP_CALLS = calls; + process.env.CODEX_GIT_DEDUP_RELEASE = release; + + try { + const first = session.requestRefresh(); + await waitForPath(calls); + const second = session.requestRefresh(); + await new Promise((resolve) => setTimeout(resolve, 100)); + expect((await readFile(calls, 'utf8')).trim().split('\n')).toHaveLength( + 1, + ); + await writeFile(release, 'continue\n'); + await Promise.all([first, second]); + } finally { + await writeFile(release, 'continue\n'); + if (previousPath === undefined) delete process.env.PATH; + else process.env.PATH = previousPath; + delete process.env.CODEX_GIT_DEDUP_CALLS; + delete process.env.CODEX_GIT_DEDUP_RELEASE; + await session.close(); + } + }); + it('closes subscriptions and rejects an in-flight read without late publication', async () => { const repository = await createRepositoryWithCommit(); const session = await createRepositoryEngine().open( @@ -208,7 +286,6 @@ describe('Repository snapshot publication', () => { done: true, value: undefined, }); - await writeFile(release, 'continue\n'); const failure: unknown = await inFlight.catch((error: unknown) => error); expect(failure).toBeInstanceOf(RepositorySessionFailure); expect(failure).toMatchObject({ @@ -424,3 +501,19 @@ async function noEvent(events: AsyncIterator): Promise { } return empty; } + +async function nextEventWithin( + events: AsyncIterator, + milliseconds: number, +): Promise> { + return Promise.race([ + events.next(), + new Promise((_, reject) => + setTimeout( + () => + reject(new Error('Timed out waiting for Repository invalidation.')), + milliseconds, + ), + ), + ]); +}