| author |  Julius Marminge <julius0216@outlook.com> 2026-04-09 11:02:43-07:00 |
| committer |  GitHub <noreply@github.com> 2026-04-09 11:02:43-07:00 |
| log | 2aa73985ea8d13d401ef9063ead4fbb9eaeeab38 |
| tree | 1db22e81387540d1176db1499b5d36d36fa3c651 |
| parent | 5b3b31b6e6846eb0509fb2e7c31646ae843c5609 |
| signature | Signed by PGP key B5690EEEBB952194 |
Refresh local git status on turn completion (#1821)
Co-authored-by: codex <codex@users.noreply.github.com>6 files changed, 131 insertions(+), 4 deletions(-)
apps
server
integration
src
orchestration
apps/server/integration/OrchestrationEngineHarness.integration.ts+17
| ... | @@ -26,6 +26,7 @@ import { CheckpointStoreLive } from "../src/checkpointing/Layers/CheckpointStore | ... | @@ -26,6 +26,7 @@ import { CheckpointStoreLive } from "../src/checkpointing/Layers/CheckpointStore |
| 26 | import { CheckpointStore } from "../src/checkpointing/Services/CheckpointStore.ts"; | 26 | import { CheckpointStore } from "../src/checkpointing/Services/CheckpointStore.ts"; |
| 27 | import { GitCoreLive } from "../src/git/Layers/GitCore.ts"; | 27 | import { GitCoreLive } from "../src/git/Layers/GitCore.ts"; |
| 28 | import { GitCore, type GitCoreShape } from "../src/git/Services/GitCore.ts"; | 28 | import { GitCore, type GitCoreShape } from "../src/git/Services/GitCore.ts"; |
| 29 | import { GitStatusBroadcaster } from "../src/git/Services/GitStatusBroadcaster.ts"; | ||
| 29 | import { TextGeneration, type TextGenerationShape } from "../src/git/Services/TextGeneration.ts"; | 30 | import { TextGeneration, type TextGenerationShape } from "../src/git/Services/TextGeneration.ts"; |
| 30 | import { OrchestrationCommandReceiptRepositoryLive } from "../src/persistence/Layers/OrchestrationCommandReceipts.ts"; | 31 | import { OrchestrationCommandReceiptRepositoryLive } from "../src/persistence/Layers/OrchestrationCommandReceipts.ts"; |
| 31 | import { OrchestrationEventStoreLive } from "../src/persistence/Layers/OrchestrationEventStore.ts"; | 32 | import { OrchestrationEventStoreLive } from "../src/persistence/Layers/OrchestrationEventStore.ts"; |
| ... | @@ -321,6 +322,22 @@ export const makeOrchestrationIntegrationHarness = ( | ... | @@ -321,6 +322,22 @@ export const makeOrchestrationIntegrationHarness = ( |
| 321 | ); | 322 | ); |
| 322 | const checkpointReactorLayer = CheckpointReactorLive.pipe( | 323 | const checkpointReactorLayer = CheckpointReactorLive.pipe( |
| 323 | Layer.provideMerge(runtimeServicesLayer), | 324 | Layer.provideMerge(runtimeServicesLayer), |
| 325 | Layer.provideMerge( | ||
| 326 | Layer.succeed(GitStatusBroadcaster, { | ||
| 327 | getStatus: () => Effect.die("getStatus should not be called in this test"), | ||
| 328 | refreshLocalStatus: () => | ||
| 329 | Effect.succeed({ | ||
| 330 | isRepo: true, | ||
| 331 | hasOriginRemote: false, | ||
| 332 | isDefaultBranch: true, | ||
| 333 | branch: "main", | ||
| 334 | hasWorkingTreeChanges: false, | ||
| 335 | workingTree: { files: [], insertions: 0, deletions: 0 }, | ||
| 336 | }), | ||
| 337 | refreshStatus: () => Effect.die("refreshStatus should not be called in this test"), | ||
| 338 | streamStatus: () => Stream.empty, | ||
| 339 | }), | ||
| 340 | ), | ||
| 324 | Layer.provideMerge( | 341 | Layer.provideMerge( |
| 325 | WorkspaceEntriesLive.pipe( | 342 | WorkspaceEntriesLive.pipe( |
| 326 | Layer.provide(WorkspacePathsLive), | 343 | Layer.provide(WorkspacePathsLive), |
apps/server/src/git/Layers/GitStatusBroadcaster.test.ts+36
| ... | @@ -143,6 +143,42 @@ describe("GitStatusBroadcasterLive", () => { | ... | @@ -143,6 +143,42 @@ describe("GitStatusBroadcasterLive", () => { |
| 143 | }).pipe(Effect.provide(makeTestLayer(state))); | 143 | }).pipe(Effect.provide(makeTestLayer(state))); |
| 144 | }); | 144 | }); |
| 145 | 145 | ||
| 146 | it.effect("refreshes only the cached local snapshot when requested", () => { | ||
| 147 | const state = { | ||
| 148 | currentLocalStatus: baseLocalStatus, | ||
| 149 | currentRemoteStatus: baseRemoteStatus, | ||
| 150 | localStatusCalls: 0, | ||
| 151 | remoteStatusCalls: 0, | ||
| 152 | localInvalidationCalls: 0, | ||
| 153 | remoteInvalidationCalls: 0, | ||
| 154 | }; | ||
| 155 | |||
| 156 | return Effect.gen(function* () { | ||
| 157 | const broadcaster = yield* GitStatusBroadcaster; | ||
| 158 | const initial = yield* broadcaster.getStatus({ cwd: "/repo" }); | ||
| 159 | |||
| 160 | state.currentLocalStatus = { | ||
| 161 | ...baseLocalStatus, | ||
| 162 | branch: "feature/local-only-refresh", | ||
| 163 | hasWorkingTreeChanges: true, | ||
| 164 | }; | ||
| 165 | |||
| 166 | const refreshedLocal = yield* broadcaster.refreshLocalStatus("/repo"); | ||
| 167 | const cached = yield* broadcaster.getStatus({ cwd: "/repo" }); | ||
| 168 | |||
| 169 | assert.deepStrictEqual(initial, baseStatus); | ||
| 170 | assert.deepStrictEqual(refreshedLocal, state.currentLocalStatus); | ||
| 171 | assert.deepStrictEqual(cached, { | ||
| 172 | ...state.currentLocalStatus, | ||
| 173 | ...baseRemoteStatus, | ||
| 174 | }); | ||
| 175 | assert.equal(state.localStatusCalls, 2); | ||
| 176 | assert.equal(state.remoteStatusCalls, 1); | ||
| 177 | assert.equal(state.localInvalidationCalls, 1); | ||
| 178 | assert.equal(state.remoteInvalidationCalls, 0); | ||
| 179 | }).pipe(Effect.provide(makeTestLayer(state))); | ||
| 180 | }); | ||
| 181 | |||
| 146 | it.effect("streams a local snapshot first and remote updates later", () => { | 182 | it.effect("streams a local snapshot first and remote updates later", () => { |
| 147 | const state = { | 183 | const state = { |
| 148 | currentLocalStatus: baseLocalStatus, | 184 | currentLocalStatus: baseLocalStatus, |
apps/server/src/git/Layers/GitStatusBroadcaster.ts+8-4
| ... | @@ -179,10 +179,13 @@ export const GitStatusBroadcasterLive = Layer.effect( | ... | @@ -179,10 +179,13 @@ export const GitStatusBroadcasterLive = Layer.effect( |
| 179 | return mergeGitStatusParts(local, remote); | 179 | return mergeGitStatusParts(local, remote); |
| 180 | }); | 180 | }); |
| 181 | 181 | ||
| 182 | const refreshLocalStatus = Effect.fn("refreshLocalStatus")(function* (cwd: string) { | 182 | const refreshLocalStatus: GitStatusBroadcasterShape["refreshLocalStatus"] = Effect.fn( |
| 183 | yield* gitManager.invalidateLocalStatus(cwd); | 183 | "refreshLocalStatus", |
| 184 | const local = yield* gitManager.localStatus({ cwd }); | 184 | )(function* (cwd) { |
| 185 | return yield* updateCachedLocalStatus(cwd, local, { publish: true }); | 185 | const normalizedCwd = normalizeCwd(cwd); |
| 186 | yield* gitManager.invalidateLocalStatus(normalizedCwd); | ||
| 187 | const local = yield* gitManager.localStatus({ cwd: normalizedCwd }); | ||
| 188 | return yield* updateCachedLocalStatus(normalizedCwd, local, { publish: true }); | ||
| 186 | }); | 189 | }); |
| 187 | 190 | ||
| 188 | const refreshRemoteStatus = Effect.fn("refreshRemoteStatus")(function* (cwd: string) { | 191 | const refreshRemoteStatus = Effect.fn("refreshRemoteStatus")(function* (cwd: string) { |
| ... | @@ -300,6 +303,7 @@ export const GitStatusBroadcasterLive = Layer.effect( | ... | @@ -300,6 +303,7 @@ export const GitStatusBroadcasterLive = Layer.effect( |
| 300 | 303 | ||
| 301 | return { | 304 | return { |
| 302 | getStatus, | 305 | getStatus, |
| 306 | refreshLocalStatus, | ||
| 303 | refreshStatus, | 307 | refreshStatus, |
| 304 | streamStatus, | 308 | streamStatus, |
| 305 | } satisfies GitStatusBroadcasterShape; | 309 | } satisfies GitStatusBroadcasterShape; |
apps/server/src/git/Services/GitStatusBroadcaster.ts+4
| ... | @@ -3,6 +3,7 @@ import type { Effect, Stream } from "effect"; | ... | @@ -3,6 +3,7 @@ import type { Effect, Stream } from "effect"; |
| 3 | import type { | 3 | import type { |
| 4 | GitManagerServiceError, | 4 | GitManagerServiceError, |
| 5 | GitStatusInput, | 5 | GitStatusInput, |
| 6 | GitStatusLocalResult, | ||
| 6 | GitStatusResult, | 7 | GitStatusResult, |
| 7 | GitStatusStreamEvent, | 8 | GitStatusStreamEvent, |
| 8 | } from "@t3tools/contracts"; | 9 | } from "@t3tools/contracts"; |
| ... | @@ -11,6 +12,9 @@ export interface GitStatusBroadcasterShape { | ... | @@ -11,6 +12,9 @@ export interface GitStatusBroadcasterShape { |
| 11 | readonly getStatus: ( | 12 | readonly getStatus: ( |
| 12 | input: GitStatusInput, | 13 | input: GitStatusInput, |
| 13 | ) => Effect.Effect<GitStatusResult, GitManagerServiceError>; | 14 | ) => Effect.Effect<GitStatusResult, GitManagerServiceError>; |
| 15 | readonly refreshLocalStatus: ( | ||
| 16 | cwd: string, | ||
| 17 | ) => Effect.Effect<GitStatusLocalResult, GitManagerServiceError>; | ||
| 14 | readonly refreshStatus: (cwd: string) => Effect.Effect<GitStatusResult, GitManagerServiceError>; | 18 | readonly refreshStatus: (cwd: string) => Effect.Effect<GitStatusResult, GitManagerServiceError>; |
| 15 | readonly streamStatus: ( | 19 | readonly streamStatus: ( |
| 16 | input: GitStatusInput, | 20 | input: GitStatusInput, |
apps/server/src/orchestration/Layers/CheckpointReactor.test.ts+43
| ... | @@ -20,6 +20,7 @@ import { afterEach, describe, expect, it, vi } from "vitest"; | ... | @@ -20,6 +20,7 @@ import { afterEach, describe, expect, it, vi } from "vitest"; |
| 20 | import { CheckpointStoreLive } from "../../checkpointing/Layers/CheckpointStore.ts"; | 20 | import { CheckpointStoreLive } from "../../checkpointing/Layers/CheckpointStore.ts"; |
| 21 | import { CheckpointStore } from "../../checkpointing/Services/CheckpointStore.ts"; | 21 | import { CheckpointStore } from "../../checkpointing/Services/CheckpointStore.ts"; |
| 22 | import { GitCoreLive } from "../../git/Layers/GitCore.ts"; | 22 | import { GitCoreLive } from "../../git/Layers/GitCore.ts"; |
| 23 | import { GitStatusBroadcaster } from "../../git/Services/GitStatusBroadcaster.ts"; | ||
| 23 | import { RepositoryIdentityResolverLive } from "../../project/Layers/RepositoryIdentityResolver.ts"; | 24 | import { RepositoryIdentityResolverLive } from "../../project/Layers/RepositoryIdentityResolver.ts"; |
| 24 | import { CheckpointReactorLive } from "./CheckpointReactor.ts"; | 25 | import { CheckpointReactorLive } from "./CheckpointReactor.ts"; |
| 25 | import { OrchestrationEngineLive } from "./OrchestrationEngine.ts"; | 26 | import { OrchestrationEngineLive } from "./OrchestrationEngine.ts"; |
| ... | @@ -243,6 +244,7 @@ describe("CheckpointReactor", () => { | ... | @@ -243,6 +244,7 @@ describe("CheckpointReactor", () => { |
| 243 | readonly threadWorktreePath?: string | null; | 244 | readonly threadWorktreePath?: string | null; |
| 244 | readonly providerSessionCwd?: string; | 245 | readonly providerSessionCwd?: string; |
| 245 | readonly providerName?: ProviderKind; | 246 | readonly providerName?: ProviderKind; |
| 247 | readonly gitStatusRefreshCalls?: Array<string>; | ||
| 246 | }) { | 248 | }) { |
| 247 | const cwd = createGitRepository(); | 249 | const cwd = createGitRepository(); |
| 248 | tempDirs.push(cwd); | 250 | tempDirs.push(cwd); |
| ... | @@ -264,11 +266,30 @@ describe("CheckpointReactor", () => { | ... | @@ -264,11 +266,30 @@ describe("CheckpointReactor", () => { |
| 264 | const ServerConfigLayer = ServerConfig.layerTest(process.cwd(), { | 266 | const ServerConfigLayer = ServerConfig.layerTest(process.cwd(), { |
| 265 | prefix: "t3-checkpoint-reactor-test-", | 267 | prefix: "t3-checkpoint-reactor-test-", |
| 266 | }); | 268 | }); |
| 269 | const gitStatusBroadcasterLayer = Layer.succeed(GitStatusBroadcaster, { | ||
| 270 | getStatus: () => Effect.die("getStatus should not be called in this test"), | ||
| 271 | refreshLocalStatus: (cwd: string) => | ||
| 272 | Effect.sync(() => { | ||
| 273 | options?.gitStatusRefreshCalls?.push(cwd); | ||
| 274 | }).pipe( | ||
| 275 | Effect.as({ | ||
| 276 | isRepo: true, | ||
| 277 | hasOriginRemote: false, | ||
| 278 | isDefaultBranch: true, | ||
| 279 | branch: "main", | ||
| 280 | hasWorkingTreeChanges: false, | ||
| 281 | workingTree: { files: [], insertions: 0, deletions: 0 }, | ||
| 282 | }), | ||
| 283 | ), | ||
| 284 | refreshStatus: () => Effect.die("refreshStatus should not be called in this test"), | ||
| 285 | streamStatus: () => Stream.empty, | ||
| 286 | }); | ||
| 267 | 287 | ||
| 268 | const layer = CheckpointReactorLive.pipe( | 288 | const layer = CheckpointReactorLive.pipe( |
| 269 | Layer.provideMerge(orchestrationLayer), | 289 | Layer.provideMerge(orchestrationLayer), |
| 270 | Layer.provideMerge(RuntimeReceiptBusLive), | 290 | Layer.provideMerge(RuntimeReceiptBusLive), |
| 271 | Layer.provideMerge(Layer.succeed(ProviderService, provider.service)), | 291 | Layer.provideMerge(Layer.succeed(ProviderService, provider.service)), |
| 292 | Layer.provideMerge(gitStatusBroadcasterLayer), | ||
| 272 | Layer.provideMerge(CheckpointStoreLive), | 293 | Layer.provideMerge(CheckpointStoreLive), |
| 273 | Layer.provideMerge(WorkspaceEntriesLive.pipe(Layer.provide(WorkspacePathsLive))), | 294 | Layer.provideMerge(WorkspaceEntriesLive.pipe(Layer.provide(WorkspacePathsLive))), |
| 274 | Layer.provideMerge(WorkspacePathsLive), | 295 | Layer.provideMerge(WorkspacePathsLive), |
| ... | @@ -426,6 +447,28 @@ describe("CheckpointReactor", () => { | ... | @@ -426,6 +447,28 @@ describe("CheckpointReactor", () => { |
| 426 | ).toBe("v2\n"); | 447 | ).toBe("v2\n"); |
| 427 | }); | 448 | }); |
| 428 | 449 | ||
| 450 | it("refreshes local git status state on turn completion using the session cwd", async () => { | ||
| 451 | const gitStatusRefreshCalls: string[] = []; | ||
| 452 | const harness = await createHarness({ | ||
| 453 | seedFilesystemCheckpoints: false, | ||
| 454 | gitStatusRefreshCalls, | ||
| 455 | }); | ||
| 456 | |||
| 457 | harness.provider.emit({ | ||
| 458 | type: "turn.completed", | ||
| 459 | eventId: EventId.makeUnsafe("evt-turn-completed-refresh-local-status"), | ||
| 460 | provider: "codex", | ||
| 461 | createdAt: new Date().toISOString(), | ||
| 462 | threadId: ThreadId.makeUnsafe("thread-1"), | ||
| 463 | turnId: asTurnId("turn-refresh-local-status"), | ||
| 464 | payload: { state: "completed" }, | ||
| 465 | }); | ||
| 466 | |||
| 467 | await harness.drain(); | ||
| 468 | |||
| 469 | expect(gitStatusRefreshCalls).toEqual([harness.cwd]); | ||
| 470 | }); | ||
| 471 | |||
| 429 | it("ignores auxiliary thread turn completion while primary turn is active", async () => { | 472 | it("ignores auxiliary thread turn completion while primary turn is active", async () => { |
| 430 | const harness = await createHarness({ seedFilesystemCheckpoints: false }); | 473 | const harness = await createHarness({ seedFilesystemCheckpoints: false }); |
| 431 | const createdAt = new Date().toISOString(); | 474 | const createdAt = new Date().toISOString(); |
apps/server/src/orchestration/Layers/CheckpointReactor.ts+23
| ... | @@ -24,6 +24,7 @@ import { RuntimeReceiptBus } from "../Services/RuntimeReceiptBus.ts"; | ... | @@ -24,6 +24,7 @@ import { RuntimeReceiptBus } from "../Services/RuntimeReceiptBus.ts"; |
| 24 | import { CheckpointStoreError } from "../../checkpointing/Errors.ts"; | 24 | import { CheckpointStoreError } from "../../checkpointing/Errors.ts"; |
| 25 | import { OrchestrationDispatchError } from "../Errors.ts"; | 25 | import { OrchestrationDispatchError } from "../Errors.ts"; |
| 26 | import { isGitRepository } from "../../git/Utils.ts"; | 26 | import { isGitRepository } from "../../git/Utils.ts"; |
| 27 | import { GitStatusBroadcaster } from "../../git/Services/GitStatusBroadcaster.ts"; | ||
| 27 | import { WorkspaceEntries } from "../../workspace/Services/WorkspaceEntries.ts"; | 28 | import { WorkspaceEntries } from "../../workspace/Services/WorkspaceEntries.ts"; |
| 28 | 29 | ||
| 29 | type ReactorInput = | 30 | type ReactorInput = |
| ... | @@ -69,6 +70,7 @@ const make = Effect.gen(function* () { | ... | @@ -69,6 +70,7 @@ const make = Effect.gen(function* () { |
| 69 | const checkpointStore = yield* CheckpointStore; | 70 | const checkpointStore = yield* CheckpointStore; |
| 70 | const receiptBus = yield* RuntimeReceiptBus; | 71 | const receiptBus = yield* RuntimeReceiptBus; |
| 71 | const workspaceEntries = yield* WorkspaceEntries; | 72 | const workspaceEntries = yield* WorkspaceEntries; |
| 73 | const gitStatusBroadcaster = yield* GitStatusBroadcaster; | ||
| 72 | 74 | ||
| 73 | const appendRevertFailureActivity = (input: { | 75 | const appendRevertFailureActivity = (input: { |
| 74 | readonly threadId: ThreadId; | 76 | readonly threadId: ThreadId; |
| ... | @@ -498,6 +500,26 @@ const make = Effect.gen(function* () { | ... | @@ -498,6 +500,26 @@ const make = Effect.gen(function* () { |
| 498 | }, | 500 | }, |
| 499 | ); | 501 | ); |
| 500 | 502 | ||
| 503 | const refreshLocalGitStatusFromTurnCompletion = Effect.fn( | ||
| 504 | "refreshLocalGitStatusFromTurnCompletion", | ||
| 505 | )(function* (event: Extract<ProviderRuntimeEvent, { type: "turn.completed" }>) { | ||
| 506 | const sessionRuntime = yield* resolveSessionRuntimeForThread(event.threadId); | ||
| 507 | if (Option.isNone(sessionRuntime)) { | ||
| 508 | return; | ||
| 509 | } | ||
| 510 | |||
| 511 | yield* gitStatusBroadcaster.refreshLocalStatus(sessionRuntime.value.cwd).pipe( | ||
| 512 | Effect.catch((error) => | ||
| 513 | Effect.logWarning("failed to refresh local git status after turn completion", { | ||
| 514 | threadId: event.threadId, | ||
| 515 | turnId: event.turnId ?? null, | ||
| 516 | cwd: sessionRuntime.value.cwd, | ||
| 517 | detail: error.message, | ||
| 518 | }), | ||
| 519 | ), | ||
| 520 | ); | ||
| 521 | }); | ||
| 522 | |||
| 501 | const ensurePreTurnBaselineFromDomainTurnStart = Effect.fn( | 523 | const ensurePreTurnBaselineFromDomainTurnStart = Effect.fn( |
| 502 | "ensurePreTurnBaselineFromDomainTurnStart", | 524 | "ensurePreTurnBaselineFromDomainTurnStart", |
| 503 | )(function* ( | 525 | )(function* ( |
| ... | @@ -736,6 +758,7 @@ const make = Effect.gen(function* () { | ... | @@ -736,6 +758,7 @@ const make = Effect.gen(function* () { |
| 736 | 758 | ||
| 737 | if (event.type === "turn.completed") { | 759 | if (event.type === "turn.completed") { |
| 738 | const turnId = toTurnId(event.turnId); | 760 | const turnId = toTurnId(event.turnId); |
| 761 | yield* refreshLocalGitStatusFromTurnCompletion(event); | ||
| 739 | yield* captureCheckpointFromTurnCompletion(event).pipe( | 762 | yield* captureCheckpointFromTurnCompletion(event).pipe( |
| 740 | Effect.catch((error) => | 763 | Effect.catch((error) => |
| 741 | appendCaptureFailureActivity({ | 764 | appendCaptureFailureActivity({ |