authorgravatar for yashafaid@gmail.comCestercian <yashafaid@gmail.com> 2026-10-02 23:52:44+05:30
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-10-02 11:22:44-07:00
loge0db2a5e58bcbe7d738bca7667d2440ddb83e30f
tree857a7a57de0c32c43e505873a92440b7f74b8912
parent5b31001eac125b501bd01decb5e04c4ff85d0c86
signature Signed by PGP key B5690EEEBB952194

fix(server): stop a second server from resending Claude turns (#13295)

2 files changed, 91 insertions(+), 1 deletions(-)

apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts+83
...@@ -1839,6 +1839,89 @@ describe("OrchestrationEngine", () => {...@@ -1839,6 +1839,89 @@ describe("OrchestrationEngine", () => {
1839 await runtime.dispose();1839 await runtime.dispose();
1840 });1840 });
18411841
1842 it("does not republish another server's turn when a local dispatch fails", async () => {
1843 const directory = await NodeFSP.mkdtemp(NodePath.join(NodeOS.tmpdir(), "t3-shared-db-"));
1844 const databasePath = NodePath.join(directory, "state.sqlite");
1845 const serverA = await createOrchestrationSystem(databasePath);
1846 const serverB = await createOrchestrationSystem(databasePath);
1847 const threadId = ThreadId.make("thread-shared");
1848 const createdAt = now();
1849 const sentinelCommandId = CommandId.make("cmd-shared-rename");
1850 try {
1851 await serverA.run(
1852 serverA.engine.dispatch({
1853 type: "project.create",
1854 commandId: CommandId.make("cmd-shared-project-create"),
1855 projectId: asProjectId("project-shared"),
1856 title: "Shared Project",
1857 workspaceRoot: "/tmp/project-shared",
1858 createdAt,
1859 }),
1860 );
1861 await serverA.run(
1862 serverA.engine.dispatch({
1863 type: "thread.create",
1864 commandId: CommandId.make("cmd-shared-thread-create"),
1865 threadId,
1866 projectId: asProjectId("project-shared"),
1867 title: "shared",
1868 modelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5-codex" },
1869 interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE,
1870 runtimeMode: "approval-required",
1871 branch: null,
1872 worktreePath: null,
1873 createdAt,
1874 }),
1875 );
1876 await serverA.run(
1877 serverA.engine.dispatch({
1878 type: "thread.turn.start",
1879 commandId: CommandId.make("cmd-shared-turn-start"),
1880 threadId,
1881 message: {
1882 messageId: asMessageId("msg-shared"),
1883 role: "user",
1884 text: "hello",
1885 attachments: [],
1886 },
1887 interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE,
1888 runtimeMode: "approval-required",
1889 createdAt,
1890 }),
1891 );
1892
1893 const published = await serverB.run(
1894 Effect.gen(function* () {
1895 const events = yield* serverB.engine.subscribeDomainEvents;
1896 // B's command model is still empty, so this fails and reconciles.
1897 yield* serverB.engine
1898 .dispatch({
1899 type: "thread.meta.update",
1900 commandId: CommandId.make("cmd-shared-stale-rename"),
1901 threadId,
1902 title: "stale",
1903 })
1904 .pipe(Effect.flip);
1905 yield* serverB.engine.dispatch({
1906 type: "thread.meta.update",
1907 commandId: sentinelCommandId,
1908 threadId,
1909 title: "renamed on B",
1910 });
1911 return yield* Stream.runCollect(
1912 Stream.takeUntil(events, (event) => event.commandId === sentinelCommandId),
1913 );
1914 }).pipe(Effect.scoped),
1915 );
1916
1917 expect(Array.from(published).map((event) => event.type)).toEqual(["thread.meta-updated"]);
1918 } finally {
1919 await serverA.dispose();
1920 await serverB.dispose();
1921 await NodeFSP.rm(directory, { recursive: true, force: true });
1922 }
1923 });
1924
1842 it("fails command dispatch when command invariants are violated", async () => {1925 it("fails command dispatch when command invariants are violated", async () => {
1843 const system = await createOrchestrationSystem();1926 const system = await createOrchestrationSystem();
1844 const { engine } = system;1927 const { engine } = system;
apps/server/src/orchestration/Layers/OrchestrationEngine.ts+8-1
...@@ -110,6 +110,10 @@ const makeOrchestrationEngine = Effect.gen(function* () {...@@ -110,6 +110,10 @@ const makeOrchestrationEngine = Effect.gen(function* () {
110110
111 const processEnvelope = (envelope: CommandEnvelope): Effect.Effect<void> => {111 const processEnvelope = (envelope: CommandEnvelope): Effect.Effect<void> => {
112 const dispatchStartSequence = commandReadModel.snapshotSequence;112 const dispatchStartSequence = commandReadModel.snapshotSequence;
113 // Events this dispatch appended. Reconcile republishes only these: a
114 // shared state directory can contain events another server already
115 // handled, and republishing them starts a second provider turn.
116 const appendedEventIds = new Set<OrchestrationEvent["eventId"]>();
113 let processingStartedAtMs = 0;117 let processingStartedAtMs = 0;
114 const aggregateRef = commandToAggregateRef(envelope.command);118 const aggregateRef = commandToAggregateRef(envelope.command);
115 const baseMetricAttributes = {119 const baseMetricAttributes = {
...@@ -127,7 +131,9 @@ const makeOrchestrationEngine = Effect.gen(function* () {...@@ -127,7 +131,9 @@ const makeOrchestrationEngine = Effect.gen(function* () {
127 commandReadModel = yield* projectEventsOntoReadModel(commandReadModel, persistedEvents);131 commandReadModel = yield* projectEventsOntoReadModel(commandReadModel, persistedEvents);
128132
129 for (const persistedEvent of persistedEvents) {133 for (const persistedEvent of persistedEvents) {
130 yield* PubSub.publish(eventPubSub, persistedEvent);134 if (appendedEventIds.has(persistedEvent.eventId)) {
135 yield* PubSub.publish(eventPubSub, persistedEvent);
136 }
131 }137 }
132 });138 });
133139
...@@ -279,6 +285,7 @@ const makeOrchestrationEngine = Effect.gen(function* () {...@@ -279,6 +285,7 @@ const makeOrchestrationEngine = Effect.gen(function* () {
279285
280 for (const nextEvent of eventBases) {286 for (const nextEvent of eventBases) {
281 const savedEvent = yield* eventStore.append(nextEvent);287 const savedEvent = yield* eventStore.append(nextEvent);
288 appendedEventIds.add(savedEvent.eventId);
282 nextCommandReadModel = yield* projectEvent(nextCommandReadModel, savedEvent);289 nextCommandReadModel = yield* projectEvent(nextCommandReadModel, savedEvent);
283 const cleanup = yield* projectionPipeline.projectEventDeferred(savedEvent);290 const cleanup = yield* projectionPipeline.projectEventDeferred(savedEvent);
284 attachmentCleanups.push(cleanup);291 attachmentCleanups.push(cleanup);