From 070e25ac5703d71c7770f4e8f3d8fb210f50bf34 Mon Sep 17 00:00:00 2001 From: Nathan Shan Date: Sun, 25 Oct 2026 12:06:17 +0800 Subject: [PATCH 01/11] =?UTF-8?q?fix(server):=20=E5=90=91=E6=8E=A5?= =?UTF-8?q?=E6=94=B6=E4=BB=A3=E7=90=86=E4=BF=9D=E7=95=99=E8=B7=A8=E7=BA=BF?= =?UTF-8?q?=E7=A8=8B=E6=B6=88=E6=81=AF=E6=9D=A5=E6=BA=90?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 在新回合和 steer 的 provider 文本中标注代理来源及发送线程 ID。 保持人类消息、无发送线程的内部消息和持久化原文不变。 通过 40 项相关测试和受影响范围检查，覆盖 SQLite 与内存投影的消息投递。 --- .../ProjectionControlReads.test.ts | 36 +++++++++++++++++-- .../ProviderTurnControlService.ts | 12 ++++--- .../ProviderTurnStartService.test.ts | 30 ++++++++++++++++ .../ProviderTurnStartService.ts | 12 ++++--- .../orchestration-v2/providerMessageOrigin.ts | 13 +++++++ 5 files changed, 93 insertions(+), 10 deletions(-) create mode 100644 apps/server/src/orchestration-v2/providerMessageOrigin.ts diff --git a/apps/server/src/orchestration-v2/ProjectionControlReads.test.ts b/apps/server/src/orchestration-v2/ProjectionControlReads.test.ts index 0f25b87f13d39..81fc420eb904c 100644 --- a/apps/server/src/orchestration-v2/ProjectionControlReads.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionControlReads.test.ts @@ -307,6 +307,32 @@ it.effect.each(storageCases)( }); yield* reply.respond({ threadId, providerSessionId, requestId, decision: "accept" }); assert.deepEqual(calls, ["interrupt", "Use the smaller fix.", "reply"]); + const messageEvent = events.find((event) => event.type === "message.updated")!; + yield* store.apply({ + ...messageEvent, + id: EventId.make("control:agent-message"), + payload: { + ...messageEvent.payload, + createdBy: "agent", + creationSource: "mcp", + senderThreadId: ThreadId.make("thread:sender"), + }, + }); + yield* control.steer({ + threadId, + providerThreadId, + providerTurnId, + providerSessionId, + messageId, + }); + const agentText = calls.at(-1); + assert.include(agentText, 'thread "thread:sender"'); + assert.include(agentText, "not a message from the human user"); + assert.isTrue(agentText?.endsWith("\n\nUse the smaller fix.")); + assert.equal( + (yield* store.getProviderControlContext(threadId, target)).message?.text, + "Use the smaller fix.", + ); const wrongSession = ProviderSessionId.make("session:wrong"); assert.equal( (yield* Effect.exit( @@ -330,7 +356,7 @@ it.effect.each(storageCases)( ))._tag, "Failure", ); - assert.lengthOf(calls, 3); + assert.lengthOf(calls, 4); const turn = context.providerTurn!; yield* store.apply({ id: EventId.make("control:turn-completed"), @@ -340,7 +366,13 @@ it.effect.each(storageCases)( payload: { ...turn, status: "completed", completedAt: now }, }); yield* control.interrupt({ threadId, providerThreadId, providerTurnId, providerSessionId }); - assert.deepEqual(calls, ["interrupt", "Use the smaller fix.", "reply", "interrupt"]); + assert.deepEqual(calls, [ + "interrupt", + "Use the smaller fix.", + "reply", + agentText, + "interrupt", + ]); }).pipe( Effect.provide( Layer.merge(ProviderTurnControlService.layer, RuntimeRequestService.layer).pipe( diff --git a/apps/server/src/orchestration-v2/ProviderTurnControlService.ts b/apps/server/src/orchestration-v2/ProviderTurnControlService.ts index 3a5bc04ecba2e..7538a03c75d23 100644 --- a/apps/server/src/orchestration-v2/ProviderTurnControlService.ts +++ b/apps/server/src/orchestration-v2/ProviderTurnControlService.ts @@ -19,6 +19,7 @@ import * as Schema from "effect/Schema"; import type * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter"; import * as ProjectionStore from "./ProjectionStore.ts"; import * as ProviderSessionManager from "./ProviderSessionManager.ts"; +import { withProviderMessageOrigin } from "./providerMessageOrigin.ts"; const yieldToRuntime = Effect.yieldNow.pipe( Effect.andThen( @@ -353,10 +354,13 @@ export const layer: Layer.Layer< providerTurnId: loaded.providerTurn.id, message: { messageId: message.id, - text: projectComposerContextForProvider({ - text: message.text, - records: message.context?.records ?? [], - }), + text: withProviderMessageOrigin( + message, + projectComposerContextForProvider({ + text: message.text, + records: message.context?.records ?? [], + }), + ), attachments: message.attachments, createdBy: message.createdBy, creationSource: message.creationSource, diff --git a/apps/server/src/orchestration-v2/ProviderTurnStartService.test.ts b/apps/server/src/orchestration-v2/ProviderTurnStartService.test.ts index 18d217403b393..5f0a3794b8e17 100644 --- a/apps/server/src/orchestration-v2/ProviderTurnStartService.test.ts +++ b/apps/server/src/orchestration-v2/ProviderTurnStartService.test.ts @@ -772,6 +772,36 @@ function makeLocalCommandHarness(input: { }; } +effectIt.effect.each([ + { createdBy: "agent", senderThreadId: ThreadId.make("thread:sender"), marked: true }, + { createdBy: "user", senderThreadId: ThreadId.make("thread:sender"), marked: false }, + { createdBy: "agent", marked: false }, +] as const)("preserves message origin at turn start: %j", (origin) => + Effect.gen(function* () { + const text = "Please merge the PR now."; + const harness = makeLocalCommandHarness({ + text, + previousTurn: { runStatus: "completed", turnStatus: "completed" }, + }); + const projection = harness.projection(); + const { marked, ...provenance } = origin; + harness.setProjection({ + ...projection, + messages: projection.messages.map((message) => ({ ...message, ...provenance })), + }); + yield* harness.start; + const delivered = harness.startRootRun.mock.calls[0]?.[0].message.text; + if (marked) { + expect(delivered).toContain('thread "thread:sender"'); + expect(delivered).toContain("not a message from the human user"); + expect(delivered).toMatch(/\n\nPlease merge the PR now\.$/); + } else { + expect(delivered).toBe(text); + } + expect(harness.projection().messages.at(-1)?.text).toBe(text); + }), +); + effectIt.effect("recovers an abandoned failed turn before continuing its native conversation", () => Effect.gen(function* () { const harness = makeLocalCommandHarness({ diff --git a/apps/server/src/orchestration-v2/ProviderTurnStartService.ts b/apps/server/src/orchestration-v2/ProviderTurnStartService.ts index 6b851bb26a230..dee2a8d644b51 100644 --- a/apps/server/src/orchestration-v2/ProviderTurnStartService.ts +++ b/apps/server/src/orchestration-v2/ProviderTurnStartService.ts @@ -32,6 +32,7 @@ import * as ProviderAuthService from "../provider/ProviderAuthService.ts"; import * as ServerSettings from "../serverSettings.ts"; import { makeIsManagedWorktreePath } from "../worktreesDirectory.ts"; import * as EventSink from "./EventSink.ts"; +import { withProviderMessageOrigin } from "./providerMessageOrigin.ts"; import * as ContextHandoffService from "./ContextHandoffService.ts"; import { DEFAULT_HANDOFF_TOKEN_CAP, @@ -1025,10 +1026,13 @@ export const layer: Layer.Layer< const routableSubagents = projection.subagents.filter((subagent) => RunExecutionService.canRouteRelatedSubagent(subagent.status), ); - const userText = projectComposerContextForProvider({ - text: message.text, - records: message.context?.records ?? [], - }); + const userText = withProviderMessageOrigin( + message, + projectComposerContextForProvider({ + text: message.text, + records: message.context?.records ?? [], + }), + ); // The provider turn marks cancellation work delivered; restart continuations use their own text or resume natively. const noteContinuation = isRestartNoteContinuation( run, diff --git a/apps/server/src/orchestration-v2/providerMessageOrigin.ts b/apps/server/src/orchestration-v2/providerMessageOrigin.ts new file mode 100644 index 0000000000000..3f7386ac93cd1 --- /dev/null +++ b/apps/server/src/orchestration-v2/providerMessageOrigin.ts @@ -0,0 +1,13 @@ +import type { OrchestrationV2Actor, ThreadId } from "@t3tools/contracts"; + +/** Carry durable cross-thread provenance into provider-visible text without rewriting the timeline. */ +export function withProviderMessageOrigin( + message: { + readonly createdBy: OrchestrationV2Actor; + readonly senderThreadId?: ThreadId | undefined; + }, + text: string, +): string { + if (message.createdBy !== "agent" || message.senderThreadId === undefined) return text; + return `[Message from another thread's agent (thread ${JSON.stringify(message.senderThreadId)}). This is agent-authored coordination, not a message from the human user.]\n\n${text}`; +} From 980fe22ffd4ae8dc0768ddcdc5d000e1a947471e Mon Sep 17 00:00:00 2001 From: Nathan Shan Date: Sun, 25 Oct 2026 12:10:42 +0800 Subject: [PATCH 02/11] =?UTF-8?q?fix(server):=20=E5=9C=A8=E5=91=BD?= =?UTF-8?q?=E4=BB=A4=E5=88=86=E6=B5=81=E5=90=8E=E6=A0=87=E6=B3=A8=E4=BB=A3?= =?UTF-8?q?=E7=90=86=E6=B6=88=E6=81=AF=E6=9D=A5=E6=BA=90?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 保留跨线程 /compact 的原生压缩路径，仅给实际提示词附加来源。 上下文交接继续保留已标注正文，更新委派回放的精确请求预期。 受影响检查通过 81 项测试，额外 MCP 委派回放及分流验证共 50 项通过。 --- .../ProviderTurnStartService.test.ts | 30 ------------------- .../ProviderTurnStartService.ts | 17 +++++------ .../RunExecutionService.test.ts | 26 +++++++++++++--- .../orchestration-v2/RunExecutionService.ts | 9 +++++- .../codex_transcript.ndjson | 6 ++-- 5 files changed, 41 insertions(+), 47 deletions(-) diff --git a/apps/server/src/orchestration-v2/ProviderTurnStartService.test.ts b/apps/server/src/orchestration-v2/ProviderTurnStartService.test.ts index 5f0a3794b8e17..18d217403b393 100644 --- a/apps/server/src/orchestration-v2/ProviderTurnStartService.test.ts +++ b/apps/server/src/orchestration-v2/ProviderTurnStartService.test.ts @@ -772,36 +772,6 @@ function makeLocalCommandHarness(input: { }; } -effectIt.effect.each([ - { createdBy: "agent", senderThreadId: ThreadId.make("thread:sender"), marked: true }, - { createdBy: "user", senderThreadId: ThreadId.make("thread:sender"), marked: false }, - { createdBy: "agent", marked: false }, -] as const)("preserves message origin at turn start: %j", (origin) => - Effect.gen(function* () { - const text = "Please merge the PR now."; - const harness = makeLocalCommandHarness({ - text, - previousTurn: { runStatus: "completed", turnStatus: "completed" }, - }); - const projection = harness.projection(); - const { marked, ...provenance } = origin; - harness.setProjection({ - ...projection, - messages: projection.messages.map((message) => ({ ...message, ...provenance })), - }); - yield* harness.start; - const delivered = harness.startRootRun.mock.calls[0]?.[0].message.text; - if (marked) { - expect(delivered).toContain('thread "thread:sender"'); - expect(delivered).toContain("not a message from the human user"); - expect(delivered).toMatch(/\n\nPlease merge the PR now\.$/); - } else { - expect(delivered).toBe(text); - } - expect(harness.projection().messages.at(-1)?.text).toBe(text); - }), -); - effectIt.effect("recovers an abandoned failed turn before continuing its native conversation", () => Effect.gen(function* () { const harness = makeLocalCommandHarness({ diff --git a/apps/server/src/orchestration-v2/ProviderTurnStartService.ts b/apps/server/src/orchestration-v2/ProviderTurnStartService.ts index dee2a8d644b51..41c2a835e5ff2 100644 --- a/apps/server/src/orchestration-v2/ProviderTurnStartService.ts +++ b/apps/server/src/orchestration-v2/ProviderTurnStartService.ts @@ -32,7 +32,6 @@ import * as ProviderAuthService from "../provider/ProviderAuthService.ts"; import * as ServerSettings from "../serverSettings.ts"; import { makeIsManagedWorktreePath } from "../worktreesDirectory.ts"; import * as EventSink from "./EventSink.ts"; -import { withProviderMessageOrigin } from "./providerMessageOrigin.ts"; import * as ContextHandoffService from "./ContextHandoffService.ts"; import { DEFAULT_HANDOFF_TOKEN_CAP, @@ -1026,13 +1025,10 @@ export const layer: Layer.Layer< const routableSubagents = projection.subagents.filter((subagent) => RunExecutionService.canRouteRelatedSubagent(subagent.status), ); - const userText = withProviderMessageOrigin( - message, - projectComposerContextForProvider({ - text: message.text, - records: message.context?.records ?? [], - }), - ); + const userText = projectComposerContextForProvider({ + text: message.text, + records: message.context?.records ?? [], + }); // The provider turn marks cancellation work delivered; restart continuations use their own text or resume natively. const noteContinuation = isRestartNoteContinuation( run, @@ -1247,7 +1243,10 @@ export const layer: Layer.Layer< ...(noteContinuation ? promptedInput : turnInput), message: { ...turnInput.message, - text: context === "" ? userText : `${context}\n\nUser message:\n${userText}`, + text: + context === "" + ? turnInput.message.text + : `${context}\n\nUser message:\n${turnInput.message.text}`, }, }).pipe( // Clear pending handoff markers for refused prompts so the next turn retains the native session. diff --git a/apps/server/src/orchestration-v2/RunExecutionService.test.ts b/apps/server/src/orchestration-v2/RunExecutionService.test.ts index 03bbbf0506b50..c0ed075212a9c 100644 --- a/apps/server/src/orchestration-v2/RunExecutionService.test.ts +++ b/apps/server/src/orchestration-v2/RunExecutionService.test.ts @@ -1028,7 +1028,13 @@ it.effect( { text: " /compact ", attachments: [], expected: "compact" }, { text: "/compact", attachments: [attachment], expected: "prompt" }, { text: "Continue the work", attachments: [], expected: "prompt" }, - ]; + ].flatMap((testCase) => + [ + { createdBy: "user", senderThreadId: ThreadId.make("thread:sender") }, + { createdBy: "agent", senderThreadId: ThreadId.make("thread:sender") }, + { createdBy: "agent" }, + ].map((origin) => ({ ...testCase, ...origin })), + ); for (const [index, testCase] of cases.entries()) { const threadId = ThreadId.make(`thread:compact-routing:${index}`); const attemptId = RunAttemptId.make(`attempt:compact-routing:${index}`); @@ -1038,12 +1044,21 @@ it.effect( providerSessionId: ProviderSessionId.make(`session:compact-routing:${index}`), session: { events: Stream.never, - startTurn: () => + startTurn: (input: ProviderAdapter.ProviderAdapterV2TurnInput) => Effect.sync(() => { + if (testCase.createdBy === "agent" && testCase.senderThreadId !== undefined) { + assert.include(input.message.text, 'thread "thread:sender"'); + assert.include(input.message.text, "not a message from the human user"); + assert.isTrue(input.message.text.endsWith(`\n\n${testCase.text}`)); + } else { + assert.equal(input.message.text, testCase.text); + } + assert.deepEqual(input.message.attachments, testCase.attachments); calls.push("prompt"); }), - compactThread: () => + compactThread: (input: ProviderAdapter.ProviderAdapterV2TurnInput) => Effect.sync(() => { + assert.equal(input.message.text, testCase.text); calls.push("compact"); }), } as unknown as ProviderAdapter.ProviderAdapterV2SessionRuntime, @@ -1070,8 +1085,11 @@ it.effect( messageId: MessageId.make(`message:compact-routing:${index}`), text: testCase.text, attachments: testCase.attachments, - createdBy: "user", + createdBy: testCase.createdBy as "user" | "agent", creationSource: "web", + ...(testCase.senderThreadId === undefined + ? {} + : { senderThreadId: testCase.senderThreadId }), }, modelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5.4" }, runtimePolicy: { diff --git a/apps/server/src/orchestration-v2/RunExecutionService.ts b/apps/server/src/orchestration-v2/RunExecutionService.ts index 9ec61cc85b994..d907dba249521 100644 --- a/apps/server/src/orchestration-v2/RunExecutionService.ts +++ b/apps/server/src/orchestration-v2/RunExecutionService.ts @@ -1,4 +1,5 @@ import { makeAssistantStreamingFilter } from "./assistantStreaming.ts"; +import { withProviderMessageOrigin } from "./providerMessageOrigin.ts"; import { resolveProjectSettings } from "@t3tools/shared/projectSettings"; import { isOrchestrationV2WorkActive, @@ -1638,7 +1639,13 @@ export const layer: Layer.Layer< cause: "This provider does not support context compaction.", }), )) - : input.session.startTurn(turnInput); + : input.session.startTurn({ + ...turnInput, + message: { + ...turnInput.message, + text: withProviderMessageOrigin(turnInput.message, turnInput.message.text), + }, + }); yield* Effect.andThen(shouldStart, startTurn).pipe( // Wait for acceptance so abandoned startup cannot leave a refresher // on the already-detached ingestion fiber. diff --git a/apps/server/src/orchestration-v2/testkit/fixtures/delegated_task_status/codex_transcript.ndjson b/apps/server/src/orchestration-v2/testkit/fixtures/delegated_task_status/codex_transcript.ndjson index 7df1add6f9c08..ab71707069a28 100644 --- a/apps/server/src/orchestration-v2/testkit/fixtures/delegated_task_status/codex_transcript.ndjson +++ b/apps/server/src/orchestration-v2/testkit/fixtures/delegated_task_status/codex_transcript.ndjson @@ -11,7 +11,7 @@ {"type":"emit_inbound","label":"parent.turn/started","frame":{"method":"turn/started","params":{"threadId":"019f1000-0000-7000-8000-000000000001","turn":{"id":"019f1000-0000-7000-8000-000000000002","items":[],"status":"inProgress","error":null,"startedAt":1776739467,"completedAt":null,"durationMs":null}}}} {"type":"expect_outbound","label":"child.thread/start","frame":{"id":4,"method":"thread/start","params":{"config":{"tools.update_plan.enabled":true,"mcp_optional_startup_grace_ms":10000}}}} {"type":"emit_inbound","label":"child.thread/start","frame":{"id":4,"result":{"thread":{"id":"019f1000-0000-7000-8000-000000000003","forkedFromId":null,"preview":"","ephemeral":false,"modelProvider":"openai","createdAt":1776739468,"updatedAt":1776739468,"status":{"type":"idle"},"path":"/home/replay-user/.codex/sessions/delegated-task-status-child.jsonl","cwd":"/home/replay-user/t3code","cliVersion":"0.120.0","source":"vscode","agentNickname":null,"agentRole":null,"gitInfo":null,"name":null,"turns":[]},"model":"gpt-5.4","modelProvider":"openai","serviceTier":"fast","cwd":"/home/replay-user/t3code","approvalPolicy":"on-request","approvalsReviewer":"user","sandbox":{"type":"workspaceWrite","writableRoots":[],"readOnlyAccess":{"type":"fullAccess"},"networkAccess":false,"excludeTmpdirEnvVar":false,"excludeSlashTmp":false},"reasoningEffort":"xhigh"}}} -{"type":"expect_outbound","label":"child.original.turn/start","frame":{"id":5,"method":"turn/start","params":{"approvalPolicy":"never","cwd":"","input":[{"text":"Inspect the delegated API boundary and return the result.","type":"text"}],"model":"gpt-5.4","sandboxPolicy":{"type":"dangerFullAccess"},"threadId":"019f1000-0000-7000-8000-000000000003","approvalsReviewer":"user","summary":"detailed"}}} +{"type":"expect_outbound","label":"child.original.turn/start","frame":{"id":5,"method":"turn/start","params":{"approvalPolicy":"never","cwd":"","input":[{"text":"[Message from another thread's agent (thread \"thread:mcp-orchestrator-parent\"). This is agent-authored coordination, not a message from the human user.]\n\nInspect the delegated API boundary and return the result.","type":"text"}],"model":"gpt-5.4","sandboxPolicy":{"type":"dangerFullAccess"},"threadId":"019f1000-0000-7000-8000-000000000003","approvalsReviewer":"user","summary":"detailed"}}} {"type":"emit_inbound","label":"child.thread/started","frame":{"method":"thread/started","params":{"thread":{"id":"019f1000-0000-7000-8000-000000000003","forkedFromId":null,"preview":"","ephemeral":false,"modelProvider":"openai","createdAt":1776739468,"updatedAt":1776739468,"status":{"type":"idle"},"path":"/home/replay-user/.codex/sessions/delegated-task-status-child.jsonl","cwd":"/home/replay-user/t3code","cliVersion":"0.120.0","source":"vscode","agentNickname":null,"agentRole":null,"gitInfo":null,"name":null,"turns":[]}}}} {"type":"emit_inbound","label":"child.original.turn/start","frame":{"id":5,"result":{"turn":{"id":"019f1000-0000-7000-8000-000000000004","items":[],"status":"inProgress","error":null,"startedAt":null,"completedAt":null,"durationMs":null}}}} {"type":"emit_inbound","label":"child.original.thread/status/changed","frame":{"method":"thread/status/changed","params":{"threadId":"019f1000-0000-7000-8000-000000000003","status":{"type":"active","activeFlags":[]}}}} @@ -21,7 +21,7 @@ {"type":"emit_inbound","label":"child.original.item/completed","frame":{"method":"item/completed","params":{"item":{"type":"agentMessage","id":"msg_delegated_task_status_original","text":"Delegated API boundary inspected.","phase":"final_answer","memoryCitation":null},"threadId":"019f1000-0000-7000-8000-000000000003","turnId":"019f1000-0000-7000-8000-000000000004"}}} {"type":"emit_inbound","label":"child.original.thread/status/changed","frame":{"method":"thread/status/changed","params":{"threadId":"019f1000-0000-7000-8000-000000000003","status":{"type":"idle"}}}} {"type":"emit_inbound","label":"child.original.turn/completed","frame":{"method":"turn/completed","params":{"threadId":"019f1000-0000-7000-8000-000000000003","turn":{"id":"019f1000-0000-7000-8000-000000000004","items":[],"status":"completed","error":null,"startedAt":1776739468,"completedAt":1776739469,"durationMs":1000}}}} -{"type":"expect_outbound","label":"child.running.turn/start","frame":{"id":6,"method":"turn/start","params":{"approvalPolicy":"never","cwd":"","input":[{"text":"Remain active until the parent cancels this delegated task.","type":"text"}],"model":"gpt-5.4","sandboxPolicy":{"type":"dangerFullAccess"},"threadId":"019f1000-0000-7000-8000-000000000003","approvalsReviewer":"user","summary":"detailed"}}} +{"type":"expect_outbound","label":"child.running.turn/start","frame":{"id":6,"method":"turn/start","params":{"approvalPolicy":"never","cwd":"","input":[{"text":"[Message from another thread's agent (thread \"thread:mcp-orchestrator-parent\"). This is agent-authored coordination, not a message from the human user.]\n\nRemain active until the parent cancels this delegated task.","type":"text"}],"model":"gpt-5.4","sandboxPolicy":{"type":"dangerFullAccess"},"threadId":"019f1000-0000-7000-8000-000000000003","approvalsReviewer":"user","summary":"detailed"}}} {"type":"emit_inbound","label":"child.running.turn/start","frame":{"id":6,"result":{"turn":{"id":"019f1000-0000-7000-8000-000000000005","items":[],"status":"inProgress","error":null,"startedAt":null,"completedAt":null,"durationMs":null}}}} {"type":"emit_inbound","label":"child.running.thread/status/changed","frame":{"method":"thread/status/changed","params":{"threadId":"019f1000-0000-7000-8000-000000000003","status":{"type":"active","activeFlags":[]}}}} {"type":"emit_inbound","label":"child.running.turn/started","frame":{"method":"turn/started","params":{"threadId":"019f1000-0000-7000-8000-000000000003","turn":{"id":"019f1000-0000-7000-8000-000000000005","items":[],"status":"inProgress","error":null,"startedAt":1776739470,"completedAt":null,"durationMs":null}}}} @@ -29,7 +29,7 @@ {"type":"emit_inbound","label":"child.running.turn/interrupt","frame":{"id":7,"result":{}}} {"type":"emit_inbound","label":"child.running.thread/status/changed","frame":{"method":"thread/status/changed","params":{"threadId":"019f1000-0000-7000-8000-000000000003","status":{"type":"idle"}}}} {"type":"emit_inbound","label":"child.running.turn/completed","frame":{"method":"turn/completed","params":{"threadId":"019f1000-0000-7000-8000-000000000003","turn":{"id":"019f1000-0000-7000-8000-000000000005","items":[],"status":"interrupted","error":null,"startedAt":1776739470,"completedAt":1776739471,"durationMs":1000}}}} -{"type":"expect_outbound","label":"child.queued.turn/start","frame":{"id":8,"method":"turn/start","params":{"approvalPolicy":"never","cwd":"","input":[{"text":"Complete the queued follow-up and return the final result.","type":"text"}],"model":"gpt-5.4","sandboxPolicy":{"type":"dangerFullAccess"},"threadId":"019f1000-0000-7000-8000-000000000003","approvalsReviewer":"user","summary":"detailed"}}} +{"type":"expect_outbound","label":"child.queued.turn/start","frame":{"id":8,"method":"turn/start","params":{"approvalPolicy":"never","cwd":"","input":[{"text":"[Message from another thread's agent (thread \"thread:mcp-orchestrator-parent\"). This is agent-authored coordination, not a message from the human user.]\n\nComplete the queued follow-up and return the final result.","type":"text"}],"model":"gpt-5.4","sandboxPolicy":{"type":"dangerFullAccess"},"threadId":"019f1000-0000-7000-8000-000000000003","approvalsReviewer":"user","summary":"detailed"}}} {"type":"emit_inbound","label":"child.queued.turn/start","frame":{"id":8,"result":{"turn":{"id":"019f1000-0000-7000-8000-000000000006","items":[],"status":"inProgress","error":null,"startedAt":null,"completedAt":null,"durationMs":null}}}} {"type":"emit_inbound","label":"child.queued.thread/status/changed","frame":{"method":"thread/status/changed","params":{"threadId":"019f1000-0000-7000-8000-000000000003","status":{"type":"active","activeFlags":[]}}}} {"type":"emit_inbound","label":"child.queued.turn/started","frame":{"method":"turn/started","params":{"threadId":"019f1000-0000-7000-8000-000000000003","turn":{"id":"019f1000-0000-7000-8000-000000000006","items":[],"status":"inProgress","error":null,"startedAt":1776739472,"completedAt":null,"durationMs":null}}}} From bedf4e81a436d84db1540d4e9adb0e2837335c2e Mon Sep 17 00:00:00 2001 From: maria-rcks Date: Sat, 31 Oct 2026 01:41:18 +0000 Subject: [PATCH 03/11] test(server): keep agent message routing coverage focused --- .../RunExecutionService.test.ts | 216 +++++++++--------- 1 file changed, 110 insertions(+), 106 deletions(-) diff --git a/apps/server/src/orchestration-v2/RunExecutionService.test.ts b/apps/server/src/orchestration-v2/RunExecutionService.test.ts index 03f5e401493d9..0c2599de65187 100644 --- a/apps/server/src/orchestration-v2/RunExecutionService.test.ts +++ b/apps/server/src/orchestration-v2/RunExecutionService.test.ts @@ -1016,115 +1016,119 @@ it.effect("fails the run when its ownership check cannot be read before calling }), ); -it.effect("preserves native command routing before marking cross-thread message origins", () => - Effect.gen(function* () { - const runExecution = yield* RunExecutionService.RunExecutionServiceV2; - const calls: Array = []; - const attachment = ChatFileAttachment.make({ - type: "file", - id: ChatAttachmentId.make("compact-file-12345678-1234-1234-1234-123456789abc"), - name: "notes.txt", - mimeType: "text/plain", - sizeBytes: 1, - }); - const cases = [ - { text: " /compact ", attachments: [], expected: "compact" }, - { text: "/compact", attachments: [attachment], expected: "prompt" }, - { text: "/goal finish the work", attachments: [], expected: "goal" }, - { text: "/goal finish the work", attachments: [attachment], expected: "prompt" }, - { text: "Continue the work", attachments: [], expected: "prompt" }, - ].flatMap((testCase) => - [ - { createdBy: "user", senderThreadId: ThreadId.make("thread:sender") }, - { createdBy: "agent", senderThreadId: ThreadId.make("thread:sender") }, - { createdBy: "agent" }, - ].map((origin) => ({ ...testCase, ...origin })), - ); - for (const [index, testCase] of cases.entries()) { - const prompts: Array = []; - const threadId = ThreadId.make(`thread:compact-routing:${index}`); - const attemptId = RunAttemptId.make(`attempt:compact-routing:${index}`); - yield* runExecution.startRootRun({ - commandId: CommandId.make(`command:compact-routing:${index}`), - appThread: rootAppThread(threadId), - providerSessionId: ProviderSessionId.make(`session:compact-routing:${index}`), - session: { - driver, - events: Stream.never, - startTurn: (input: ProviderAdapter.ProviderAdapterV2TurnInput) => - Effect.sync(() => { - prompts.push(input); - calls.push(testCase.expected === "goal" ? "goal" : "prompt"); - }), - compactThread: (input: ProviderAdapter.ProviderAdapterV2TurnInput) => - Effect.sync(() => { - assert.equal(input.message.text, testCase.text); - calls.push("compact"); - }), - } as unknown as ProviderAdapter.ProviderAdapterV2SessionRuntime, - run: { - id: RunId.make(`run:compact-routing:${index}`), - threadId, - ordinal: 1, - providerInstanceId: ProviderInstanceId.make("codex"), - } as OrchestrationV2Run, - rootNode: { - id: NodeId.make(`node:compact-routing:${index}`), - } as OrchestrationV2ExecutionNode, - checkpointScope: { - id: CheckpointScopeId.make(`checkpoint:compact-routing:${index}`), - } as OrchestrationV2CheckpointScope, - providerThread: { - id: ProviderThreadId.make(`provider-thread:compact-routing:${index}`), - driver, - } as OrchestrationV2ProviderThread, - attempt: { id: attemptId, providerTurnId: null } as OrchestrationV2RunAttempt, - attemptId, - providerTurnOrdinal: 1, - message: { - messageId: MessageId.make(`message:compact-routing:${index}`), - text: testCase.text, - attachments: testCase.attachments, - createdBy: testCase.createdBy as "user" | "agent", - creationSource: "web", - ...(testCase.senderThreadId === undefined - ? {} - : { senderThreadId: testCase.senderThreadId }), - }, - modelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5.4" }, - runtimePolicy: { - runtimeMode: "full-access", - interactionMode: "default", - cwd: process.cwd(), - approvalPolicy: "never", - sandboxPolicy: { - type: "readOnly", - access: { type: "fullAccess" }, - networkAccess: false, - }, - }, +it.effect( + "dispatches only attachment-free compact commands through the native compaction path", + () => + Effect.gen(function* () { + const runExecution = yield* RunExecutionService.RunExecutionServiceV2; + const calls: Array = []; + const attachment = ChatFileAttachment.make({ + type: "file", + id: ChatAttachmentId.make("compact-file-12345678-1234-1234-1234-123456789abc"), + name: "notes.txt", + mimeType: "text/plain", + sizeBytes: 1, }); - const prompt = prompts[0]; - if (prompt !== undefined) { - if ( - testCase.expected !== "goal" && - testCase.createdBy === "agent" && - testCase.senderThreadId !== undefined - ) { - assert.include(prompt.message.text, 'thread "thread:sender"'); - assert.include(prompt.message.text, "not a message from the human user"); - assert.isTrue(prompt.message.text.endsWith(`\n\n${testCase.text}`)); - } else { - assert.equal(prompt.message.text, testCase.text); + const cases = [ + { text: " /compact ", attachments: [], expected: "compact" }, + { text: "/compact", attachments: [attachment], expected: "prompt" }, + { text: "/goal finish the work", attachments: [], expected: "goal" }, + { text: "/goal finish the work", attachments: [attachment], expected: "prompt" }, + { text: "Continue the work", attachments: [], expected: "prompt" }, + ].flatMap((testCase) => + ( + [ + { createdBy: "user", senderThreadId: ThreadId.make("thread:sender") }, + { createdBy: "agent", senderThreadId: ThreadId.make("thread:sender") }, + { createdBy: "agent", senderThreadId: undefined }, + ] as const + ).map((origin) => ({ ...testCase, ...origin })), + ); + for (const [index, testCase] of cases.entries()) { + const prompts: Array = []; + const threadId = ThreadId.make(`thread:compact-routing:${index}`); + const attemptId = RunAttemptId.make(`attempt:compact-routing:${index}`); + yield* runExecution.startRootRun({ + commandId: CommandId.make(`command:compact-routing:${index}`), + appThread: rootAppThread(threadId), + providerSessionId: ProviderSessionId.make(`session:compact-routing:${index}`), + session: { + driver, + events: Stream.never, + startTurn: (input: ProviderAdapter.ProviderAdapterV2TurnInput) => + Effect.sync(() => { + prompts.push(input); + calls.push("prompt"); + }), + compactThread: (input: ProviderAdapter.ProviderAdapterV2TurnInput) => + Effect.sync(() => { + assert.equal(input.message.text, testCase.text); + calls.push("compact"); + }), + } as unknown as ProviderAdapter.ProviderAdapterV2SessionRuntime, + run: { + id: RunId.make(`run:compact-routing:${index}`), + threadId, + ordinal: 1, + providerInstanceId: ProviderInstanceId.make("codex"), + } as OrchestrationV2Run, + rootNode: { + id: NodeId.make(`node:compact-routing:${index}`), + } as OrchestrationV2ExecutionNode, + checkpointScope: { + id: CheckpointScopeId.make(`checkpoint:compact-routing:${index}`), + } as OrchestrationV2CheckpointScope, + providerThread: { + id: ProviderThreadId.make(`provider-thread:compact-routing:${index}`), + driver, + } as OrchestrationV2ProviderThread, + attempt: { id: attemptId, providerTurnId: null } as OrchestrationV2RunAttempt, + attemptId, + providerTurnOrdinal: 1, + message: { + messageId: MessageId.make(`message:compact-routing:${index}`), + text: testCase.text, + attachments: testCase.attachments, + createdBy: testCase.createdBy, + creationSource: "web", + ...(testCase.senderThreadId === undefined + ? {} + : { senderThreadId: testCase.senderThreadId }), + }, + modelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5.4" }, + runtimePolicy: { + runtimeMode: "full-access", + interactionMode: "default", + cwd: process.cwd(), + approvalPolicy: "never", + sandboxPolicy: { + type: "readOnly", + access: { type: "fullAccess" }, + networkAccess: false, + }, + }, + }); + const prompt = prompts[0]; + if (prompt !== undefined) { + if ( + testCase.expected !== "goal" && + testCase.createdBy === "agent" && + testCase.senderThreadId !== undefined + ) { + assert.include(prompt.message.text, 'thread "thread:sender"'); + assert.include(prompt.message.text, "not a message from the human user"); + assert.isTrue(prompt.message.text.endsWith(`\n\n${testCase.text}`)); + } else { + assert.equal(prompt.message.text, testCase.text); + } + assert.deepEqual(prompt.message.attachments, testCase.attachments); } - assert.deepEqual(prompt.message.attachments, testCase.attachments); } - } - assert.deepEqual( - calls, - cases.map((testCase) => testCase.expected), - ); - }).pipe(Effect.provide(layerRunExecutionTest)), + assert.deepEqual( + calls, + cases.map((testCase) => (testCase.expected === "compact" ? "compact" : "prompt")), + ); + }).pipe(Effect.provide(layerRunExecutionTest)), ); it.effect("starts the provider when checkpoint baseline capture fails", () => From dc47cb839758f0e3a102ee4bdf26ebe8c47003f6 Mon Sep 17 00:00:00 2001 From: maria-rcks Date: Sat, 31 Oct 2026 01:49:34 +0000 Subject: [PATCH 04/11] fix(server): preserve native commands when marking agent messages --- .../ProjectionControlReads.test.ts | 24 ++++++++- .../ProviderTurnControlService.ts | 1 + .../RunExecutionService.test.ts | 49 ++++++++++++++----- .../orchestration-v2/RunExecutionService.ts | 12 ++--- .../orchestration-v2/providerMessageOrigin.ts | 13 ++--- 5 files changed, 74 insertions(+), 25 deletions(-) diff --git a/apps/server/src/orchestration-v2/ProjectionControlReads.test.ts b/apps/server/src/orchestration-v2/ProjectionControlReads.test.ts index 132756f0773cd..43be8affadb52 100644 --- a/apps/server/src/orchestration-v2/ProjectionControlReads.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionControlReads.test.ts @@ -275,6 +275,7 @@ it.effect.each(storageCases)( const layerSessions = Layer.mock(ProviderSessionManager.ProviderSessionManagerV2)({ get: () => Effect.succeedSome({ + driver: ProviderDriverKind.make("pi"), interruptTurn: () => Effect.sync(() => { calls.push("interrupt"); @@ -334,6 +335,26 @@ it.effect.each(storageCases)( (yield* store.getProviderControlContext(threadId, target)).message?.text, "Use the smaller fix.", ); + const compactText = "/compact keep the tests"; + yield* store.apply({ + ...messageEvent, + id: EventId.make("control:agent-compact"), + payload: { + ...messageEvent.payload, + text: compactText, + createdBy: "agent", + creationSource: "mcp", + senderThreadId: ThreadId.make("thread:sender"), + }, + }); + yield* control.steer({ + threadId, + providerThreadId, + providerTurnId, + providerSessionId, + messageId, + }); + assert.equal(calls.at(-1), compactText); const wrongSession = ProviderSessionId.make("session:wrong"); assert.equal( (yield* Effect.exit( @@ -357,7 +378,7 @@ it.effect.each(storageCases)( ))._tag, "Failure", ); - assert.lengthOf(calls, 4); + assert.lengthOf(calls, 5); const turn = context.providerTurn!; yield* store.apply({ id: EventId.make("control:turn-completed"), @@ -372,6 +393,7 @@ it.effect.each(storageCases)( "Use the smaller fix.", "reply", agentText, + compactText, "interrupt", ]); }).pipe( diff --git a/apps/server/src/orchestration-v2/ProviderTurnControlService.ts b/apps/server/src/orchestration-v2/ProviderTurnControlService.ts index a423ad9e9aa73..44b7c8e788109 100644 --- a/apps/server/src/orchestration-v2/ProviderTurnControlService.ts +++ b/apps/server/src/orchestration-v2/ProviderTurnControlService.ts @@ -381,6 +381,7 @@ export const layer: Layer.Layer< text: message.text, records: message.context?.records ?? [], }), + loaded.session.value.driver, ), attachments: message.attachments, createdBy: message.createdBy, diff --git a/apps/server/src/orchestration-v2/RunExecutionService.test.ts b/apps/server/src/orchestration-v2/RunExecutionService.test.ts index 0c2599de65187..a7f1de17d8407 100644 --- a/apps/server/src/orchestration-v2/RunExecutionService.test.ts +++ b/apps/server/src/orchestration-v2/RunExecutionService.test.ts @@ -1031,19 +1031,45 @@ it.effect( }); const cases = [ { text: " /compact ", attachments: [], expected: "compact" }, + { + text: "/compact keep the tests", + attachments: [], + expected: "command", + driver: ProviderDriverKind.make("pi"), + }, + { + text: "/compact keep the tests", + attachments: [attachment], + expected: "command", + driver: ProviderDriverKind.make("pi"), + }, + { + text: "Continue the work", + attachments: [], + expected: "prompt", + driver: ProviderDriverKind.make("pi"), + }, + { + text: "/review", + attachments: [], + expected: "command", + driver: ProviderDriverKind.make("opencode"), + }, { text: "/compact", attachments: [attachment], expected: "prompt" }, { text: "/goal finish the work", attachments: [], expected: "goal" }, { text: "/goal finish the work", attachments: [attachment], expected: "prompt" }, { text: "Continue the work", attachments: [], expected: "prompt" }, - ].flatMap((testCase) => - ( - [ - { createdBy: "user", senderThreadId: ThreadId.make("thread:sender") }, - { createdBy: "agent", senderThreadId: ThreadId.make("thread:sender") }, - { createdBy: "agent", senderThreadId: undefined }, - ] as const - ).map((origin) => ({ ...testCase, ...origin })), - ); + ] + .map((testCase) => ({ driver, ...testCase })) + .flatMap((testCase) => + ( + [ + { createdBy: "user", senderThreadId: ThreadId.make("thread:sender") }, + { createdBy: "agent", senderThreadId: ThreadId.make("thread:sender") }, + { createdBy: "agent", senderThreadId: undefined }, + ] as const + ).map((origin) => ({ ...testCase, ...origin })), + ); for (const [index, testCase] of cases.entries()) { const prompts: Array = []; const threadId = ThreadId.make(`thread:compact-routing:${index}`); @@ -1053,7 +1079,7 @@ it.effect( appThread: rootAppThread(threadId), providerSessionId: ProviderSessionId.make(`session:compact-routing:${index}`), session: { - driver, + driver: testCase.driver, events: Stream.never, startTurn: (input: ProviderAdapter.ProviderAdapterV2TurnInput) => Effect.sync(() => { @@ -1080,7 +1106,7 @@ it.effect( } as OrchestrationV2CheckpointScope, providerThread: { id: ProviderThreadId.make(`provider-thread:compact-routing:${index}`), - driver, + driver: testCase.driver, } as OrchestrationV2ProviderThread, attempt: { id: attemptId, providerTurnId: null } as OrchestrationV2RunAttempt, attemptId, @@ -1112,6 +1138,7 @@ it.effect( if (prompt !== undefined) { if ( testCase.expected !== "goal" && + testCase.expected !== "command" && testCase.createdBy === "agent" && testCase.senderThreadId !== undefined ) { diff --git a/apps/server/src/orchestration-v2/RunExecutionService.ts b/apps/server/src/orchestration-v2/RunExecutionService.ts index df140b56c1444..82317ac19455b 100644 --- a/apps/server/src/orchestration-v2/RunExecutionService.ts +++ b/apps/server/src/orchestration-v2/RunExecutionService.ts @@ -3,7 +3,6 @@ import { withProviderMessageOrigin } from "./providerMessageOrigin.ts"; import { resolveProjectSettings } from "@t3tools/shared/projectSettings"; import { isOrchestrationV2WorkActive, - parseGoalCommand, CommandId, type EventId, type ModelSelection, @@ -1679,9 +1678,6 @@ export const layer: Layer.Layer< const compact = input.message.attachments.length === 0 && input.message.text.trim().toLowerCase() === "/compact"; - const goalCommand = - (input.session.driver === "codex" || input.session.driver === "claudeAgent") && - parseGoalCommand(input.message) !== null; const startTurn = compact ? (input.session.compactThread?.(turnInput) ?? Effect.fail( @@ -1697,9 +1693,11 @@ export const layer: Layer.Layer< ...turnInput, message: { ...turnInput.message, - text: goalCommand - ? turnInput.message.text - : withProviderMessageOrigin(turnInput.message, turnInput.message.text), + text: withProviderMessageOrigin( + turnInput.message, + turnInput.message.text, + input.session.driver, + ), }, }); yield* Effect.andThen(shouldStart, startTurn).pipe( diff --git a/apps/server/src/orchestration-v2/providerMessageOrigin.ts b/apps/server/src/orchestration-v2/providerMessageOrigin.ts index 3f7386ac93cd1..e4c0148f64c1e 100644 --- a/apps/server/src/orchestration-v2/providerMessageOrigin.ts +++ b/apps/server/src/orchestration-v2/providerMessageOrigin.ts @@ -1,13 +1,14 @@ -import type { OrchestrationV2Actor, ThreadId } from "@t3tools/contracts"; +import type { ProviderDriverKind } from "@t3tools/contracts"; +import type { ProviderAdapterV2TurnMessage } from "@t3tools/provider-core/server/ProviderAdapter"; -/** Carry durable cross-thread provenance into provider-visible text without rewriting the timeline. */ export function withProviderMessageOrigin( - message: { - readonly createdBy: OrchestrationV2Actor; - readonly senderThreadId?: ThreadId | undefined; - }, + message: Pick, text: string, + driver: ProviderDriverKind, ): string { if (message.createdBy !== "agent" || message.senderThreadId === undefined) return text; + // Native commands need their leading slash; Pi parses commands even with attachments. + if (text.trimStart().startsWith("/") && (message.attachments.length === 0 || driver === "pi")) + return text; return `[Message from another thread's agent (thread ${JSON.stringify(message.senderThreadId)}). This is agent-authored coordination, not a message from the human user.]\n\n${text}`; } From 62bd769932006e84ab4016375a06170ae6df6c86 Mon Sep 17 00:00:00 2001 From: maria-rcks Date: Sat, 31 Oct 2026 01:53:13 +0000 Subject: [PATCH 05/11] fix(server): keep attached OpenCode commands parseable --- .../ProjectionControlReads.test.ts | 33 ++++++++++++++----- .../RunExecutionService.test.ts | 2 +- .../orchestration-v2/providerMessageOrigin.ts | 7 ++-- 3 files changed, 31 insertions(+), 11 deletions(-) diff --git a/apps/server/src/orchestration-v2/ProjectionControlReads.test.ts b/apps/server/src/orchestration-v2/ProjectionControlReads.test.ts index 43be8affadb52..569f8f0c01991 100644 --- a/apps/server/src/orchestration-v2/ProjectionControlReads.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionControlReads.test.ts @@ -214,9 +214,18 @@ it.effect.each(storageCases)( assert.isUndefined((yield* store.getRunningTurnContext(threadId)).providerTurn); }).pipe(Effect.provide(storeLayer)), ); -it.effect.each(storageCases)( - "$storage: controls and replies read only their exact durable targets", - ({ storage, storeLayer }) => +it.effect.each( + storageCases.flatMap((storageCase) => [ + { + ...storageCase, + commandDriver: ProviderDriverKind.make("pi"), + commandText: "/compact keep the tests", + }, + { ...storageCase, commandDriver: ProviderDriverKind.make("opencode"), commandText: "/review" }, + ]), +)( + "$storage: controls and replies read only their exact durable targets ($commandDriver)", + ({ storage, storeLayer, commandDriver, commandText }) => Effect.gen(function* () { const store = yield* ProjectionStore.ProjectionStoreV2; const now = yield* DateTime.now; @@ -275,7 +284,7 @@ it.effect.each(storageCases)( const layerSessions = Layer.mock(ProviderSessionManager.ProviderSessionManagerV2)({ get: () => Effect.succeedSome({ - driver: ProviderDriverKind.make("pi"), + driver: commandDriver, interruptTurn: () => Effect.sync(() => { calls.push("interrupt"); @@ -335,13 +344,21 @@ it.effect.each(storageCases)( (yield* store.getProviderControlContext(threadId, target)).message?.text, "Use the smaller fix.", ); - const compactText = "/compact keep the tests"; yield* store.apply({ ...messageEvent, id: EventId.make("control:agent-compact"), payload: { ...messageEvent.payload, - text: compactText, + text: commandText, + attachments: [ + { + type: "file", + id: "command-attachment", + name: "notes.txt", + mimeType: "text/plain", + sizeBytes: 1, + }, + ], createdBy: "agent", creationSource: "mcp", senderThreadId: ThreadId.make("thread:sender"), @@ -354,7 +371,7 @@ it.effect.each(storageCases)( providerSessionId, messageId, }); - assert.equal(calls.at(-1), compactText); + assert.equal(calls.at(-1), commandText); const wrongSession = ProviderSessionId.make("session:wrong"); assert.equal( (yield* Effect.exit( @@ -393,7 +410,7 @@ it.effect.each(storageCases)( "Use the smaller fix.", "reply", agentText, - compactText, + commandText, "interrupt", ]); }).pipe( diff --git a/apps/server/src/orchestration-v2/RunExecutionService.test.ts b/apps/server/src/orchestration-v2/RunExecutionService.test.ts index a7f1de17d8407..1029d77fa4887 100644 --- a/apps/server/src/orchestration-v2/RunExecutionService.test.ts +++ b/apps/server/src/orchestration-v2/RunExecutionService.test.ts @@ -1051,7 +1051,7 @@ it.effect( }, { text: "/review", - attachments: [], + attachments: [attachment], expected: "command", driver: ProviderDriverKind.make("opencode"), }, diff --git a/apps/server/src/orchestration-v2/providerMessageOrigin.ts b/apps/server/src/orchestration-v2/providerMessageOrigin.ts index e4c0148f64c1e..92c8cd26430f7 100644 --- a/apps/server/src/orchestration-v2/providerMessageOrigin.ts +++ b/apps/server/src/orchestration-v2/providerMessageOrigin.ts @@ -7,8 +7,11 @@ export function withProviderMessageOrigin( driver: ProviderDriverKind, ): string { if (message.createdBy !== "agent" || message.senderThreadId === undefined) return text; - // Native commands need their leading slash; Pi parses commands even with attachments. - if (text.trimStart().startsWith("/") && (message.attachments.length === 0 || driver === "pi")) + // Native commands need their leading slash; Pi and OpenCode also parse them with attachments. + if ( + text.trimStart().startsWith("/") && + (message.attachments.length === 0 || driver === "pi" || driver === "opencode") + ) return text; return `[Message from another thread's agent (thread ${JSON.stringify(message.senderThreadId)}). This is agent-authored coordination, not a message from the human user.]\n\n${text}`; } From 187051d6633e1036caecda931c400b4a3f5bb439 Mon Sep 17 00:00:00 2001 From: maria-rcks Date: Sat, 31 Oct 2026 02:02:36 +0000 Subject: [PATCH 06/11] fix(server): keep provenance on slash-prefixed agent prose --- .../ProjectionControlReads.test.ts | 31 ++++++++++++------- .../RunExecutionService.test.ts | 13 +++++++- .../orchestration-v2/RunExecutionService.ts | 16 +++++++--- .../orchestration-v2/providerMessageOrigin.ts | 9 ++---- 4 files changed, 45 insertions(+), 24 deletions(-) diff --git a/apps/server/src/orchestration-v2/ProjectionControlReads.test.ts b/apps/server/src/orchestration-v2/ProjectionControlReads.test.ts index 569f8f0c01991..5127d8c5b2181 100644 --- a/apps/server/src/orchestration-v2/ProjectionControlReads.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionControlReads.test.ts @@ -222,6 +222,7 @@ it.effect.each( commandText: "/compact keep the tests", }, { ...storageCase, commandDriver: ProviderDriverKind.make("opencode"), commandText: "/review" }, + { ...storageCase, commandDriver: driver, commandText: "/goal finish the work" }, ]), )( "$storage: controls and replies read only their exact durable targets ($commandDriver)", @@ -350,15 +351,18 @@ it.effect.each( payload: { ...messageEvent.payload, text: commandText, - attachments: [ - { - type: "file", - id: "command-attachment", - name: "notes.txt", - mimeType: "text/plain", - sizeBytes: 1, - }, - ], + attachments: + commandDriver === "codex" + ? [] + : [ + { + type: "file", + id: "command-attachment", + name: "notes.txt", + mimeType: "text/plain", + sizeBytes: 1, + }, + ], createdBy: "agent", creationSource: "mcp", senderThreadId: ThreadId.make("thread:sender"), @@ -371,7 +375,12 @@ it.effect.each( providerSessionId, messageId, }); - assert.equal(calls.at(-1), commandText); + const deliveredCommandText = calls.at(-1); + if (commandDriver === "pi") assert.equal(deliveredCommandText, commandText); + else { + assert.include(deliveredCommandText, "not a message from the human user"); + assert.isTrue(deliveredCommandText?.endsWith(`\n\n${commandText}`)); + } const wrongSession = ProviderSessionId.make("session:wrong"); assert.equal( (yield* Effect.exit( @@ -410,7 +419,7 @@ it.effect.each( "Use the smaller fix.", "reply", agentText, - commandText, + deliveredCommandText, "interrupt", ]); }).pipe( diff --git a/apps/server/src/orchestration-v2/RunExecutionService.test.ts b/apps/server/src/orchestration-v2/RunExecutionService.test.ts index 1029d77fa4887..3f26a1f82645d 100644 --- a/apps/server/src/orchestration-v2/RunExecutionService.test.ts +++ b/apps/server/src/orchestration-v2/RunExecutionService.test.ts @@ -1031,6 +1031,11 @@ it.effect( }); const cases = [ { text: " /compact ", attachments: [], expected: "compact" }, + { + text: "/workspace/src/main.ts is the file to update.", + attachments: [], + expected: "prompt", + }, { text: "/compact keep the tests", attachments: [], @@ -1052,7 +1057,13 @@ it.effect( { text: "/review", attachments: [attachment], - expected: "command", + expected: "prompt", + driver: ProviderDriverKind.make("opencode"), + }, + { + text: "/README.md is the file to update.", + attachments: [], + expected: "prompt", driver: ProviderDriverKind.make("opencode"), }, { text: "/compact", attachments: [attachment], expected: "prompt" }, diff --git a/apps/server/src/orchestration-v2/RunExecutionService.ts b/apps/server/src/orchestration-v2/RunExecutionService.ts index 82317ac19455b..2127605fb7d19 100644 --- a/apps/server/src/orchestration-v2/RunExecutionService.ts +++ b/apps/server/src/orchestration-v2/RunExecutionService.ts @@ -3,6 +3,7 @@ import { withProviderMessageOrigin } from "./providerMessageOrigin.ts"; import { resolveProjectSettings } from "@t3tools/shared/projectSettings"; import { isOrchestrationV2WorkActive, + parseGoalCommand, CommandId, type EventId, type ModelSelection, @@ -1678,6 +1679,9 @@ export const layer: Layer.Layer< const compact = input.message.attachments.length === 0 && input.message.text.trim().toLowerCase() === "/compact"; + const goalCommand = + (input.session.driver === "codex" || input.session.driver === "claudeAgent") && + parseGoalCommand(input.message) !== null; const startTurn = compact ? (input.session.compactThread?.(turnInput) ?? Effect.fail( @@ -1693,11 +1697,13 @@ export const layer: Layer.Layer< ...turnInput, message: { ...turnInput.message, - text: withProviderMessageOrigin( - turnInput.message, - turnInput.message.text, - input.session.driver, - ), + text: goalCommand + ? turnInput.message.text + : withProviderMessageOrigin( + turnInput.message, + turnInput.message.text, + input.session.driver, + ), }, }); yield* Effect.andThen(shouldStart, startTurn).pipe( diff --git a/apps/server/src/orchestration-v2/providerMessageOrigin.ts b/apps/server/src/orchestration-v2/providerMessageOrigin.ts index 92c8cd26430f7..fc3323fcb7abb 100644 --- a/apps/server/src/orchestration-v2/providerMessageOrigin.ts +++ b/apps/server/src/orchestration-v2/providerMessageOrigin.ts @@ -2,16 +2,11 @@ import type { ProviderDriverKind } from "@t3tools/contracts"; import type { ProviderAdapterV2TurnMessage } from "@t3tools/provider-core/server/ProviderAdapter"; export function withProviderMessageOrigin( - message: Pick, + message: Pick, text: string, driver: ProviderDriverKind, ): string { if (message.createdBy !== "agent" || message.senderThreadId === undefined) return text; - // Native commands need their leading slash; Pi and OpenCode also parse them with attachments. - if ( - text.trimStart().startsWith("/") && - (message.attachments.length === 0 || driver === "pi" || driver === "opencode") - ) - return text; + if (driver === "pi" && /^\/compact(?:\s|$)/u.test(text.trim())) return text; return `[Message from another thread's agent (thread ${JSON.stringify(message.senderThreadId)}). This is agent-authored coordination, not a message from the human user.]\n\n${text}`; } From cd17a5f8190c7d5ae363bc16e53a0207076d93c3 Mon Sep 17 00:00:00 2001 From: Nathan Shan Date: Sat, 7 Nov 2026 21:43:27 +0800 Subject: [PATCH 07/11] =?UTF-8?q?fix(server):=20=E5=8E=9F=E7=94=9F?= =?UTF-8?q?=E5=91=BD=E4=BB=A4=E6=AD=A3=E6=96=87=E4=BF=9D=E7=95=99=E8=B7=A8?= =?UTF-8?q?=E7=BA=BF=E7=A8=8B=E6=B6=88=E6=81=AF=E6=9D=A5=E6=BA=90?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 在原生目标正文和 Pi 压缩指令中保留发送线程及 agent 来源标记。 - 保持命令前缀、目标控制操作及人工消息原文不变。 - 扩展现有启动与引导回归，变更范围检查和 158 项测试通过。 --- .../ProjectionControlReads.test.ts | 8 ++++-- .../RunExecutionService.test.ts | 28 +++++++++++++++++-- .../orchestration-v2/RunExecutionService.ts | 28 +++++++++++++------ .../orchestration-v2/providerMessageOrigin.ts | 9 ++++-- 4 files changed, 57 insertions(+), 16 deletions(-) diff --git a/apps/server/src/orchestration-v2/ProjectionControlReads.test.ts b/apps/server/src/orchestration-v2/ProjectionControlReads.test.ts index 5127d8c5b2181..00398f462173e 100644 --- a/apps/server/src/orchestration-v2/ProjectionControlReads.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionControlReads.test.ts @@ -376,8 +376,12 @@ it.effect.each( messageId, }); const deliveredCommandText = calls.at(-1); - if (commandDriver === "pi") assert.equal(deliveredCommandText, commandText); - else { + if (commandDriver === "pi") { + assert.isTrue(deliveredCommandText?.startsWith("/compact [Message from")); + assert.include(deliveredCommandText, 'thread "thread:sender"'); + assert.include(deliveredCommandText, "not a message from the human user"); + assert.isTrue(deliveredCommandText?.endsWith("\n\nkeep the tests")); + } else { assert.include(deliveredCommandText, "not a message from the human user"); assert.isTrue(deliveredCommandText?.endsWith(`\n\n${commandText}`)); } diff --git a/apps/server/src/orchestration-v2/RunExecutionService.test.ts b/apps/server/src/orchestration-v2/RunExecutionService.test.ts index 3f76b51d0594b..7f55586582378 100644 --- a/apps/server/src/orchestration-v2/RunExecutionService.test.ts +++ b/apps/server/src/orchestration-v2/RunExecutionService.test.ts @@ -1,5 +1,6 @@ import { assert, it } from "@effect/vitest"; import { + parseGoalCommand, CheckpointScopeId, ChatAttachmentId, ChatFileAttachment, @@ -1083,6 +1084,18 @@ it.effect( }, { text: "/compact", attachments: [attachment], expected: "prompt" }, { text: "/goal finish the work", attachments: [], expected: "goal" }, + { text: " /goal finish the work ", attachments: [], expected: "goal" }, + { + text: "/goal finish the work", + attachments: [], + expected: "goal", + driver: ProviderDriverKind.make("claudeAgent"), + }, + ...["/goal", "/goal edit", "/goal clear", "/goal pause", "/goal resume"].map((text) => ({ + text, + attachments: [], + expected: "control", + })), { text: "/goal finish the work", attachments: [attachment], expected: "prompt" }, { text: "Continue the work", attachments: [], expected: "prompt" }, ] @@ -1163,14 +1176,23 @@ it.effect( const prompt = prompts[0]; if (prompt !== undefined) { if ( - testCase.expected !== "goal" && - testCase.expected !== "command" && + testCase.expected !== "control" && testCase.createdBy === "agent" && testCase.senderThreadId !== undefined ) { assert.include(prompt.message.text, 'thread "thread:sender"'); assert.include(prompt.message.text, "not a message from the human user"); - assert.isTrue(prompt.message.text.endsWith(`\n\n${testCase.text}`)); + if (testCase.expected === "goal") { + const goal = parseGoalCommand(prompt.message); + assert.equal(goal?.type, "set"); + if (goal?.type === "set") + assert.isTrue(goal.objective.endsWith("\n\nfinish the work")); + } else if (testCase.expected === "command") { + assert.isTrue(prompt.message.text.startsWith("/compact [Message from")); + assert.isTrue(prompt.message.text.endsWith("\n\nkeep the tests")); + } else { + assert.isTrue(prompt.message.text.endsWith(`\n\n${testCase.text}`)); + } } else { assert.equal(prompt.message.text, testCase.text); } diff --git a/apps/server/src/orchestration-v2/RunExecutionService.ts b/apps/server/src/orchestration-v2/RunExecutionService.ts index b6caf0f61666e..1f4227ef5167d 100644 --- a/apps/server/src/orchestration-v2/RunExecutionService.ts +++ b/apps/server/src/orchestration-v2/RunExecutionService.ts @@ -1730,8 +1730,9 @@ export const layer: Layer.Layer< input.message.attachments.length === 0 && input.message.text.trim().toLowerCase() === "/compact"; const goalCommand = - (input.session.driver === "codex" || input.session.driver === "claudeAgent") && - parseGoalCommand(input.message) !== null; + input.session.driver === "codex" || input.session.driver === "claudeAgent" + ? parseGoalCommand(input.message) + : null; const startTurn = compact ? (input.session.compactThread?.(turnInput) ?? Effect.fail( @@ -1747,13 +1748,22 @@ export const layer: Layer.Layer< ...turnInput, message: { ...turnInput.message, - text: goalCommand - ? turnInput.message.text - : withProviderMessageOrigin( - turnInput.message, - turnInput.message.text, - input.session.driver, - ), + text: + goalCommand?.type === "set" && + turnInput.message.createdBy === "agent" && + turnInput.message.senderThreadId !== undefined + ? `/goal ${withProviderMessageOrigin( + turnInput.message, + goalCommand.objective, + input.session.driver, + )}` + : goalCommand !== null + ? turnInput.message.text + : withProviderMessageOrigin( + turnInput.message, + turnInput.message.text, + input.session.driver, + ), }, }); yield* Effect.andThen(shouldStart, startTurn).pipe( diff --git a/apps/server/src/orchestration-v2/providerMessageOrigin.ts b/apps/server/src/orchestration-v2/providerMessageOrigin.ts index fc3323fcb7abb..c898e063c45bb 100644 --- a/apps/server/src/orchestration-v2/providerMessageOrigin.ts +++ b/apps/server/src/orchestration-v2/providerMessageOrigin.ts @@ -7,6 +7,11 @@ export function withProviderMessageOrigin( driver: ProviderDriverKind, ): string { if (message.createdBy !== "agent" || message.senderThreadId === undefined) return text; - if (driver === "pi" && /^\/compact(?:\s|$)/u.test(text.trim())) return text; - return `[Message from another thread's agent (thread ${JSON.stringify(message.senderThreadId)}). This is agent-authored coordination, not a message from the human user.]\n\n${text}`; + const heading = `[Message from another thread's agent (thread ${JSON.stringify(message.senderThreadId)}). This is agent-authored coordination, not a message from the human user.]`; + const compact = driver === "pi" ? /^\/compact(?:\s+([\s\S]*))?$/u.exec(text.trim()) : null; + if (compact !== null) { + const instructions = compact[1]?.trim(); + return instructions ? `/compact ${heading}\n\n${instructions}` : text; + } + return `${heading}\n\n${text}`; } From 3ed09f81d66608e34e76f47c2d07029bcd2b1e0e Mon Sep 17 00:00:00 2001 From: Nathan Shan Date: Thu, 12 Nov 2026 10:32:11 +0800 Subject: [PATCH 08/11] =?UTF-8?q?fix(server):=20=E4=BF=9D=E7=95=99=20OpenC?= =?UTF-8?q?ode=20=E5=8E=9F=E7=94=9F=E7=9B=AE=E6=A0=87=E5=91=BD=E4=BB=A4?= =?UTF-8?q?=E7=9A=84=E6=89=A7=E8=A1=8C=E8=AF=AD=E4=B9=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 按会话能力识别支持原生目标的 OpenCode，保留目标命令格式和消息来源。 - 补充目标设置、清除和旧版普通消息路径的回归覆盖。 - 变更范围检查通过，包含六个文件的 169 项测试。 --- .../RunExecutionService.test.ts | 353 ++++++++++-------- .../orchestration-v2/RunExecutionService.ts | 10 +- 2 files changed, 193 insertions(+), 170 deletions(-) diff --git a/apps/server/src/orchestration-v2/RunExecutionService.test.ts b/apps/server/src/orchestration-v2/RunExecutionService.test.ts index 51778b5f907e7..f171ccff98871 100644 --- a/apps/server/src/orchestration-v2/RunExecutionService.test.ts +++ b/apps/server/src/orchestration-v2/RunExecutionService.test.ts @@ -1039,178 +1039,199 @@ it.effect("fails the run when its ownership check cannot be read before calling }), ); -it.effect( - "dispatches only attachment-free compact commands through the native compaction path", - () => - Effect.gen(function* () { - const runExecution = yield* RunExecutionService.RunExecutionServiceV2; - const calls: Array = []; - const attachment = ChatFileAttachment.make({ - type: "file", - id: ChatAttachmentId.make("compact-file-12345678-1234-1234-1234-123456789abc"), - name: "notes.txt", - mimeType: "text/plain", - sizeBytes: 1, - }); - const cases = [ - { text: " /compact ", attachments: [], expected: "compact" }, - { - text: "/workspace/src/main.ts is the file to update.", - attachments: [], - expected: "prompt", - }, - { - text: "/compact keep the tests", - attachments: [], - expected: "command", - driver: ProviderDriverKind.make("pi"), - }, - { - text: "/compact keep the tests", - attachments: [attachment], - expected: "command", - driver: ProviderDriverKind.make("pi"), - }, - { - text: "Continue the work", - attachments: [], - expected: "prompt", - driver: ProviderDriverKind.make("pi"), - }, - { - text: "/review", - attachments: [attachment], - expected: "prompt", - driver: ProviderDriverKind.make("opencode"), - }, - { - text: "/README.md is the file to update.", - attachments: [], - expected: "prompt", - driver: ProviderDriverKind.make("opencode"), - }, - { text: "/compact", attachments: [attachment], expected: "prompt" }, - { text: "/goal finish the work", attachments: [], expected: "goal" }, - { text: " /goal finish the work ", attachments: [], expected: "goal" }, - { - text: "/goal finish the work", - attachments: [], - expected: "goal", - driver: ProviderDriverKind.make("claudeAgent"), - }, - ...["/goal", "/goal edit", "/goal clear", "/goal pause", "/goal resume"].map((text) => ({ - text, - attachments: [], - expected: "control", - })), - { text: "/goal finish the work", attachments: [attachment], expected: "prompt" }, - { text: "Continue the work", attachments: [], expected: "prompt" }, - ] - .map((testCase) => ({ driver, ...testCase })) - .flatMap((testCase) => - ( - [ - { createdBy: "user", senderThreadId: ThreadId.make("thread:sender") }, - { createdBy: "agent", senderThreadId: ThreadId.make("thread:sender") }, - { createdBy: "agent", senderThreadId: undefined }, - ] as const - ).map((origin) => ({ ...testCase, ...origin })), - ); - for (const [index, testCase] of cases.entries()) { - const prompts: Array = []; - const threadId = ThreadId.make(`thread:compact-routing:${index}`); - const attemptId = RunAttemptId.make(`attempt:compact-routing:${index}`); - yield* runExecution.startRootRun({ - commandId: CommandId.make(`command:compact-routing:${index}`), - appThread: rootAppThread(threadId), - providerSessionId: ProviderSessionId.make(`session:compact-routing:${index}`), - session: { - driver: testCase.driver, - events: Stream.never, - startTurn: (input: ProviderAdapter.ProviderAdapterV2TurnInput) => - Effect.sync(() => { - prompts.push(input); - calls.push("prompt"); - }), - compactThread: (input: ProviderAdapter.ProviderAdapterV2TurnInput) => - Effect.sync(() => { - assert.equal(input.message.text, testCase.text); - calls.push("compact"); - }), - } as unknown as ProviderAdapter.ProviderAdapterV2SessionRuntime, - run: { - id: RunId.make(`run:compact-routing:${index}`), - threadId, - ordinal: 1, - providerInstanceId: ProviderInstanceId.make("codex"), - } as OrchestrationV2Run, - rootNode: { - id: NodeId.make(`node:compact-routing:${index}`), - } as OrchestrationV2ExecutionNode, - checkpointScope: { - id: CheckpointScopeId.make(`checkpoint:compact-routing:${index}`), - } as OrchestrationV2CheckpointScope, - providerThread: { - id: ProviderThreadId.make(`provider-thread:compact-routing:${index}`), - driver: testCase.driver, - } as OrchestrationV2ProviderThread, - attempt: { id: attemptId, providerTurnId: null } as OrchestrationV2RunAttempt, - attemptId, - providerTurnOrdinal: 1, - message: { - messageId: MessageId.make(`message:compact-routing:${index}`), - text: testCase.text, - attachments: testCase.attachments, - createdBy: testCase.createdBy, - creationSource: "web", - ...(testCase.senderThreadId === undefined - ? {} - : { senderThreadId: testCase.senderThreadId }), +it.effect("preserves native compaction and goal command routing when labeling agent messages", () => + Effect.gen(function* () { + const runExecution = yield* RunExecutionService.RunExecutionServiceV2; + const calls: Array = []; + const attachment = ChatFileAttachment.make({ + type: "file", + id: ChatAttachmentId.make("compact-file-12345678-1234-1234-1234-123456789abc"), + name: "notes.txt", + mimeType: "text/plain", + sizeBytes: 1, + }); + const cases = [ + { text: " /compact ", attachments: [], expected: "compact" }, + { + text: "/workspace/src/main.ts is the file to update.", + attachments: [], + expected: "prompt", + }, + { + text: "/compact keep the tests", + attachments: [], + expected: "command", + driver: ProviderDriverKind.make("pi"), + }, + { + text: "/compact keep the tests", + attachments: [attachment], + expected: "command", + driver: ProviderDriverKind.make("pi"), + }, + { + text: "Continue the work", + attachments: [], + expected: "prompt", + driver: ProviderDriverKind.make("pi"), + }, + { + text: "/review", + attachments: [attachment], + expected: "prompt", + driver: ProviderDriverKind.make("opencode"), + }, + { + text: "/README.md is the file to update.", + attachments: [], + expected: "prompt", + driver: ProviderDriverKind.make("opencode"), + }, + { text: "/compact", attachments: [attachment], expected: "prompt" }, + { text: "/goal finish the work", attachments: [], expected: "goal" }, + { text: " /goal finish the work ", attachments: [], expected: "goal" }, + { + text: "/goal finish the work", + attachments: [], + expected: "goal", + driver: ProviderDriverKind.make("claudeAgent"), + }, + { + text: "/goal finish the work", + attachments: [], + expected: "goal", + driver: ProviderDriverKind.make("opencode"), + nativeGoals: true, + }, + { + text: "/goal clear", + attachments: [], + expected: "control", + driver: ProviderDriverKind.make("opencode"), + nativeGoals: true, + }, + { + text: "/goal finish the work", + attachments: [], + expected: "prompt", + driver: ProviderDriverKind.make("opencode"), + nativeGoals: false, + }, + ...["/goal", "/goal edit", "/goal clear", "/goal pause", "/goal resume"].map((text) => ({ + text, + attachments: [], + expected: "control", + })), + { text: "/goal finish the work", attachments: [attachment], expected: "prompt" }, + { text: "Continue the work", attachments: [], expected: "prompt" }, + ] + .map((testCase) => ({ driver, ...testCase })) + .flatMap((testCase) => + ( + [ + { createdBy: "user", senderThreadId: ThreadId.make("thread:sender") }, + { createdBy: "agent", senderThreadId: ThreadId.make("thread:sender") }, + { createdBy: "agent", senderThreadId: undefined }, + ] as const + ).map((origin) => ({ ...testCase, ...origin })), + ); + for (const [index, testCase] of cases.entries()) { + const prompts: Array = []; + const threadId = ThreadId.make(`thread:compact-routing:${index}`); + const attemptId = RunAttemptId.make(`attempt:compact-routing:${index}`); + yield* runExecution.startRootRun({ + commandId: CommandId.make(`command:compact-routing:${index}`), + appThread: rootAppThread(threadId), + providerSessionId: ProviderSessionId.make(`session:compact-routing:${index}`), + session: { + driver: testCase.driver, + providerSession: { + capabilities: { turns: { supportsNativeGoals: testCase.nativeGoals } }, }, - modelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5.4" }, - runtimePolicy: { - runtimeMode: "full-access", - interactionMode: "default", - cwd: process.cwd(), - approvalPolicy: "never", - sandboxPolicy: { - type: "readOnly", - access: { type: "fullAccess" }, - networkAccess: false, - }, + events: Stream.never, + startTurn: (input: ProviderAdapter.ProviderAdapterV2TurnInput) => + Effect.sync(() => { + prompts.push(input); + calls.push("prompt"); + }), + compactThread: (input: ProviderAdapter.ProviderAdapterV2TurnInput) => + Effect.sync(() => { + assert.equal(input.message.text, testCase.text); + calls.push("compact"); + }), + } as unknown as ProviderAdapter.ProviderAdapterV2SessionRuntime, + run: { + id: RunId.make(`run:compact-routing:${index}`), + threadId, + ordinal: 1, + providerInstanceId: ProviderInstanceId.make("codex"), + } as OrchestrationV2Run, + rootNode: { + id: NodeId.make(`node:compact-routing:${index}`), + } as OrchestrationV2ExecutionNode, + checkpointScope: { + id: CheckpointScopeId.make(`checkpoint:compact-routing:${index}`), + } as OrchestrationV2CheckpointScope, + providerThread: { + id: ProviderThreadId.make(`provider-thread:compact-routing:${index}`), + driver: testCase.driver, + } as OrchestrationV2ProviderThread, + attempt: { id: attemptId, providerTurnId: null } as OrchestrationV2RunAttempt, + attemptId, + providerTurnOrdinal: 1, + message: { + messageId: MessageId.make(`message:compact-routing:${index}`), + text: testCase.text, + attachments: testCase.attachments, + createdBy: testCase.createdBy, + creationSource: "web", + ...(testCase.senderThreadId === undefined + ? {} + : { senderThreadId: testCase.senderThreadId }), + }, + modelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5.4" }, + runtimePolicy: { + runtimeMode: "full-access", + interactionMode: "default", + cwd: process.cwd(), + approvalPolicy: "never", + sandboxPolicy: { + type: "readOnly", + access: { type: "fullAccess" }, + networkAccess: false, }, - }); - const prompt = prompts[0]; - if (prompt !== undefined) { - if ( - testCase.expected !== "control" && - testCase.createdBy === "agent" && - testCase.senderThreadId !== undefined - ) { - assert.include(prompt.message.text, 'thread "thread:sender"'); - assert.include(prompt.message.text, "not a message from the human user"); - if (testCase.expected === "goal") { - const goal = parseGoalCommand(prompt.message); - assert.equal(goal?.type, "set"); - if (goal?.type === "set") - assert.isTrue(goal.objective.endsWith("\n\nfinish the work")); - } else if (testCase.expected === "command") { - assert.isTrue(prompt.message.text.startsWith("/compact [Message from")); - assert.isTrue(prompt.message.text.endsWith("\n\nkeep the tests")); - } else { - assert.isTrue(prompt.message.text.endsWith(`\n\n${testCase.text}`)); - } + }, + }); + const prompt = prompts[0]; + if (prompt !== undefined) { + if ( + testCase.expected !== "control" && + testCase.createdBy === "agent" && + testCase.senderThreadId !== undefined + ) { + assert.include(prompt.message.text, 'thread "thread:sender"'); + assert.include(prompt.message.text, "not a message from the human user"); + if (testCase.expected === "goal") { + const goal = parseGoalCommand(prompt.message); + assert.equal(goal?.type, "set"); + if (goal?.type === "set") assert.isTrue(goal.objective.endsWith("\n\nfinish the work")); + } else if (testCase.expected === "command") { + assert.isTrue(prompt.message.text.startsWith("/compact [Message from")); + assert.isTrue(prompt.message.text.endsWith("\n\nkeep the tests")); } else { - assert.equal(prompt.message.text, testCase.text); + assert.isTrue(prompt.message.text.endsWith(`\n\n${testCase.text}`)); } - assert.deepEqual(prompt.message.attachments, testCase.attachments); + } else { + assert.equal(prompt.message.text, testCase.text); } + assert.deepEqual(prompt.message.attachments, testCase.attachments); } - assert.deepEqual( - calls, - cases.map((testCase) => (testCase.expected === "compact" ? "compact" : "prompt")), - ); - }).pipe(Effect.provide(layerRunExecutionTest)), + } + assert.deepEqual( + calls, + cases.map((testCase) => (testCase.expected === "compact" ? "compact" : "prompt")), + ); + }).pipe(Effect.provide(layerRunExecutionTest)), ); it.effect("starts the provider when checkpoint baseline capture fails", () => diff --git a/apps/server/src/orchestration-v2/RunExecutionService.ts b/apps/server/src/orchestration-v2/RunExecutionService.ts index f269206fa72c8..a9e9324cc61a1 100644 --- a/apps/server/src/orchestration-v2/RunExecutionService.ts +++ b/apps/server/src/orchestration-v2/RunExecutionService.ts @@ -1819,10 +1819,12 @@ export const layer: Layer.Layer< const compact = input.message.attachments.length === 0 && input.message.text.trim().toLowerCase() === "/compact"; - const goalCommand = - input.session.driver === "codex" || input.session.driver === "claudeAgent" - ? parseGoalCommand(input.message) - : null; + const goalDriver = + input.session.driver === "codex" || + input.session.driver === "claudeAgent" || + (input.session.driver === "opencode" && + input.session.providerSession.capabilities.turns.supportsNativeGoals === true); + const goalCommand = goalDriver ? parseGoalCommand(input.message) : null; const startTurn = compact ? (input.session.compactThread?.(turnInput) ?? Effect.fail( From fdd0618968daae021cec23a440e95c980e9dbc4c Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Thu, 12 Nov 2026 22:02:30 +0000 Subject: [PATCH 09/11] fix(server): keep native goal routing capability-based for agent messages Share one native-goal predicate between run execution and turn start so cross-thread agent goals on capability-declaring drivers such as Antigravity keep `/goal` first, and only read session capabilities for cross-thread agent messages. Cut the routing tests to explicit cases. AI-assisted; verified with the tests and tools listed above. Co-Authored-By: Claude Opus 5.5 --- .../ProjectionControlReads.test.ts | 61 +------ .../ProviderTurnStartService.ts | 14 +- .../RunExecutionService.test.ts | 149 +++++------------- .../orchestration-v2/RunExecutionService.ts | 51 +++--- .../orchestration-v2/providerMessageOrigin.ts | 9 +- 5 files changed, 81 insertions(+), 203 deletions(-) diff --git a/apps/server/src/orchestration-v2/ProjectionControlReads.test.ts b/apps/server/src/orchestration-v2/ProjectionControlReads.test.ts index 00398f462173e..2a03aed74319a 100644 --- a/apps/server/src/orchestration-v2/ProjectionControlReads.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionControlReads.test.ts @@ -214,19 +214,9 @@ it.effect.each(storageCases)( assert.isUndefined((yield* store.getRunningTurnContext(threadId)).providerTurn); }).pipe(Effect.provide(storeLayer)), ); -it.effect.each( - storageCases.flatMap((storageCase) => [ - { - ...storageCase, - commandDriver: ProviderDriverKind.make("pi"), - commandText: "/compact keep the tests", - }, - { ...storageCase, commandDriver: ProviderDriverKind.make("opencode"), commandText: "/review" }, - { ...storageCase, commandDriver: driver, commandText: "/goal finish the work" }, - ]), -)( - "$storage: controls and replies read only their exact durable targets ($commandDriver)", - ({ storage, storeLayer, commandDriver, commandText }) => +it.effect.each(storageCases)( + "$storage: controls and replies read only their exact durable targets", + ({ storage, storeLayer }) => Effect.gen(function* () { const store = yield* ProjectionStore.ProjectionStoreV2; const now = yield* DateTime.now; @@ -285,7 +275,7 @@ it.effect.each( const layerSessions = Layer.mock(ProviderSessionManager.ProviderSessionManagerV2)({ get: () => Effect.succeedSome({ - driver: commandDriver, + driver, interruptTurn: () => Effect.sync(() => { calls.push("interrupt"); @@ -345,46 +335,6 @@ it.effect.each( (yield* store.getProviderControlContext(threadId, target)).message?.text, "Use the smaller fix.", ); - yield* store.apply({ - ...messageEvent, - id: EventId.make("control:agent-compact"), - payload: { - ...messageEvent.payload, - text: commandText, - attachments: - commandDriver === "codex" - ? [] - : [ - { - type: "file", - id: "command-attachment", - name: "notes.txt", - mimeType: "text/plain", - sizeBytes: 1, - }, - ], - createdBy: "agent", - creationSource: "mcp", - senderThreadId: ThreadId.make("thread:sender"), - }, - }); - yield* control.steer({ - threadId, - providerThreadId, - providerTurnId, - providerSessionId, - messageId, - }); - const deliveredCommandText = calls.at(-1); - if (commandDriver === "pi") { - assert.isTrue(deliveredCommandText?.startsWith("/compact [Message from")); - assert.include(deliveredCommandText, 'thread "thread:sender"'); - assert.include(deliveredCommandText, "not a message from the human user"); - assert.isTrue(deliveredCommandText?.endsWith("\n\nkeep the tests")); - } else { - assert.include(deliveredCommandText, "not a message from the human user"); - assert.isTrue(deliveredCommandText?.endsWith(`\n\n${commandText}`)); - } const wrongSession = ProviderSessionId.make("session:wrong"); assert.equal( (yield* Effect.exit( @@ -408,7 +358,7 @@ it.effect.each( ))._tag, "Failure", ); - assert.lengthOf(calls, 5); + assert.lengthOf(calls, 4); const turn = context.providerTurn!; yield* store.apply({ id: EventId.make("control:turn-completed"), @@ -423,7 +373,6 @@ it.effect.each( "Use the smaller fix.", "reply", agentText, - deliveredCommandText, "interrupt", ]); }).pipe( diff --git a/apps/server/src/orchestration-v2/ProviderTurnStartService.ts b/apps/server/src/orchestration-v2/ProviderTurnStartService.ts index ca27bb3ec9367..1b84692675e1f 100644 --- a/apps/server/src/orchestration-v2/ProviderTurnStartService.ts +++ b/apps/server/src/orchestration-v2/ProviderTurnStartService.ts @@ -61,6 +61,7 @@ import { makeProviderFailure, } from "@t3tools/provider-core/server/failure"; import * as RunExecutionService from "./RunExecutionService.ts"; +import { isCrossThreadAgentMessage } from "./providerMessageOrigin.ts"; import * as RuntimePolicy from "./RuntimePolicy.ts"; import * as TurnItemPositionStore from "./TurnItemPositionStore.ts"; import { @@ -870,10 +871,7 @@ export const layer: Layer.Layer< records: message.context?.records ?? [], }); // Native goal commands go verbatim; handoff history waits a turn. - const goalDriver = - session.driver === "codex" || - session.driver === "claudeAgent" || - session.providerSession.capabilities.turns.supportsNativeGoals === true; + const goalDriver = RunExecutionService.acceptsNativeGoalCommands(session); const goalCommand = goalDriver && parseGoalCommand(message) !== null; // The provider turn marks cancellation work delivered; restart continuations use their own text or resume natively. const noteContinuation = isRestartNoteContinuation( @@ -1756,11 +1754,9 @@ export const layer: Layer.Layer< .filter((part) => part !== "") .join("\n\n"); const { restartContinuationOfRunId: _resumedRunId, ...promptedInput } = turnInput; - const messageHeading = - turnInput.message.createdBy === "agent" && - turnInput.message.senderThreadId !== undefined - ? "Agent message:" - : "User message:"; + const messageHeading = isCrossThreadAgentMessage(turnInput.message) + ? "Agent message:" + : "User message:"; yield* start({ ...(noteContinuation ? promptedInput : turnInput), message: { diff --git a/apps/server/src/orchestration-v2/RunExecutionService.test.ts b/apps/server/src/orchestration-v2/RunExecutionService.test.ts index 22a28e2dc2551..6c95bed7388e1 100644 --- a/apps/server/src/orchestration-v2/RunExecutionService.test.ts +++ b/apps/server/src/orchestration-v2/RunExecutionService.test.ts @@ -1,6 +1,5 @@ import { assert, it } from "@effect/vitest"; import { - parseGoalCommand, CheckpointScopeId, ChatAttachmentId, ChatFileAttachment, @@ -1043,7 +1042,7 @@ it.effect("fails the run when its ownership check cannot be read before calling }), ); -it.effect("preserves native compaction and goal command routing when labeling agent messages", () => +it.effect("routes compaction and goal commands and marks cross-thread agent messages", () => Effect.gen(function* () { const runExecution = yield* RunExecutionService.RunExecutionServiceV2; const calls: Array = []; @@ -1054,113 +1053,67 @@ it.effect("preserves native compaction and goal command routing when labeling ag mimeType: "text/plain", sizeBytes: 1, }); - const cases = [ - { text: " /compact ", attachments: [], expected: "compact" }, + const origin = + '[Message from another thread\'s agent (thread "thread:sender").' + + " This is agent-authored coordination, not a message from the human user.]"; + const agent = { createdBy: "agent", senderThreadId: ThreadId.make("thread:sender") } as const; + const cases: ReadonlyArray<{ + readonly text: string; + readonly expected: string; + readonly attachments?: ReadonlyArray; + readonly createdBy?: "user" | "agent"; + readonly senderThreadId?: ThreadId; + readonly driver?: ProviderDriverKind; + readonly nativeGoals?: boolean; + }> = [ + { text: " /compact ", expected: "compact" }, + { text: "/compact", attachments: [attachment], expected: "/compact" }, + { text: "Continue the work", expected: "Continue the work" }, + { createdBy: "agent", text: "Continue the work", expected: "Continue the work" }, + { ...agent, text: "Continue the work", expected: `${origin}\n\nContinue the work` }, { - text: "/workspace/src/main.ts is the file to update.", - attachments: [], - expected: "prompt", - }, - { - text: "/compact keep the tests", - attachments: [], - expected: "command", + ...agent, driver: ProviderDriverKind.make("pi"), - }, - { text: "/compact keep the tests", - attachments: [attachment], - expected: "command", - driver: ProviderDriverKind.make("pi"), - }, - { - text: "Continue the work", - attachments: [], - expected: "prompt", - driver: ProviderDriverKind.make("pi"), - }, - { - text: "/review", - attachments: [attachment], - expected: "prompt", - driver: ProviderDriverKind.make("opencode"), + expected: `/compact ${origin}\n\nkeep the tests`, }, + { ...agent, text: "/goal finish", expected: `/goal ${origin}\n\nfinish` }, { - text: "/README.md is the file to update.", - attachments: [], - expected: "prompt", - driver: ProviderDriverKind.make("opencode"), - }, - { text: "/compact", attachments: [attachment], expected: "prompt" }, - { text: "/goal finish the work", attachments: [], expected: "goal" }, - { text: " /goal finish the work ", attachments: [], expected: "goal" }, - { - text: "/goal finish the work", - attachments: [], - expected: "goal", - driver: ProviderDriverKind.make("claudeAgent"), - }, - { - text: "/goal finish the work", - attachments: [], - expected: "goal", - driver: ProviderDriverKind.make("opencode"), - nativeGoals: true, - }, - { - text: "/goal clear", - attachments: [], - expected: "control", - driver: ProviderDriverKind.make("opencode"), + ...agent, + driver: ProviderDriverKind.make("antigravity"), nativeGoals: true, + text: "/goal finish", + expected: `/goal ${origin}\n\nfinish`, }, { - text: "/goal finish the work", - attachments: [], - expected: "prompt", + ...agent, driver: ProviderDriverKind.make("opencode"), nativeGoals: false, + text: "/goal finish", + expected: `${origin}\n\n/goal finish`, }, - ...["/goal", "/goal edit", "/goal clear", "/goal pause", "/goal resume"].map((text) => ({ - text, - attachments: [], - expected: "control", - })), - { text: "/goal finish the work", attachments: [attachment], expected: "prompt" }, - { text: "Continue the work", attachments: [], expected: "prompt" }, - ] - .map((testCase) => ({ driver, ...testCase })) - .flatMap((testCase) => - ( - [ - { createdBy: "user", senderThreadId: ThreadId.make("thread:sender") }, - { createdBy: "agent", senderThreadId: ThreadId.make("thread:sender") }, - { createdBy: "agent", senderThreadId: undefined }, - ] as const - ).map((origin) => ({ ...testCase, ...origin })), - ); + { ...agent, text: "/goal clear", expected: "/goal clear" }, + ]; for (const [index, testCase] of cases.entries()) { - const prompts: Array = []; const threadId = ThreadId.make(`thread:compact-routing:${index}`); const attemptId = RunAttemptId.make(`attempt:compact-routing:${index}`); + const caseDriver = testCase.driver ?? driver; yield* runExecution.startRootRun({ commandId: CommandId.make(`command:compact-routing:${index}`), appThread: rootAppThread(threadId), providerSessionId: ProviderSessionId.make(`session:compact-routing:${index}`), session: { - driver: testCase.driver, + driver: caseDriver, providerSession: { capabilities: { turns: { supportsNativeGoals: testCase.nativeGoals } }, }, events: Stream.never, startTurn: (input: ProviderAdapter.ProviderAdapterV2TurnInput) => Effect.sync(() => { - prompts.push(input); - calls.push("prompt"); + calls.push(input.message.text); }), - compactThread: (input: ProviderAdapter.ProviderAdapterV2TurnInput) => + compactThread: () => Effect.sync(() => { - assert.equal(input.message.text, testCase.text); calls.push("compact"); }), } as unknown as ProviderAdapter.ProviderAdapterV2SessionRuntime, @@ -1178,7 +1131,7 @@ it.effect("preserves native compaction and goal command routing when labeling ag } as OrchestrationV2CheckpointScope, providerThread: { id: ProviderThreadId.make(`provider-thread:compact-routing:${index}`), - driver: testCase.driver, + driver: caseDriver, } as OrchestrationV2ProviderThread, attempt: { id: attemptId, providerTurnId: null } as OrchestrationV2RunAttempt, attemptId, @@ -1186,8 +1139,8 @@ it.effect("preserves native compaction and goal command routing when labeling ag message: { messageId: MessageId.make(`message:compact-routing:${index}`), text: testCase.text, - attachments: testCase.attachments, - createdBy: testCase.createdBy, + attachments: testCase.attachments ?? [], + createdBy: testCase.createdBy ?? "user", creationSource: "web", ...(testCase.senderThreadId === undefined ? {} @@ -1206,34 +1159,10 @@ it.effect("preserves native compaction and goal command routing when labeling ag }, }, }); - const prompt = prompts[0]; - if (prompt !== undefined) { - if ( - testCase.expected !== "control" && - testCase.createdBy === "agent" && - testCase.senderThreadId !== undefined - ) { - assert.include(prompt.message.text, 'thread "thread:sender"'); - assert.include(prompt.message.text, "not a message from the human user"); - if (testCase.expected === "goal") { - const goal = parseGoalCommand(prompt.message); - assert.equal(goal?.type, "set"); - if (goal?.type === "set") assert.isTrue(goal.objective.endsWith("\n\nfinish the work")); - } else if (testCase.expected === "command") { - assert.isTrue(prompt.message.text.startsWith("/compact [Message from")); - assert.isTrue(prompt.message.text.endsWith("\n\nkeep the tests")); - } else { - assert.isTrue(prompt.message.text.endsWith(`\n\n${testCase.text}`)); - } - } else { - assert.equal(prompt.message.text, testCase.text); - } - assert.deepEqual(prompt.message.attachments, testCase.attachments); - } } assert.deepEqual( calls, - cases.map((testCase) => (testCase.expected === "compact" ? "compact" : "prompt")), + cases.map((testCase) => testCase.expected), ); }).pipe(Effect.provide(layerRunExecutionTest)), ); diff --git a/apps/server/src/orchestration-v2/RunExecutionService.ts b/apps/server/src/orchestration-v2/RunExecutionService.ts index 83d3bcb62a76d..f673d98420a21 100644 --- a/apps/server/src/orchestration-v2/RunExecutionService.ts +++ b/apps/server/src/orchestration-v2/RunExecutionService.ts @@ -1,5 +1,5 @@ import { makeAssistantStreamingFilter } from "./assistantStreaming.ts"; -import { withProviderMessageOrigin } from "./providerMessageOrigin.ts"; +import { isCrossThreadAgentMessage, withProviderMessageOrigin } from "./providerMessageOrigin.ts"; import { resolveProjectSettings } from "@t3tools/shared/projectSettings"; import { isOrchestrationV2WorkActive, @@ -190,6 +190,16 @@ function childTurnsContinue(terminal: ProviderTerminalEvent): boolean { } /** Launching runs still ingest running subagents; taking them over would duplicate events. */ +export function acceptsNativeGoalCommands( + session: Pick, +): boolean { + return ( + session.driver === "codex" || + session.driver === "claudeAgent" || + session.providerSession.capabilities.turns.supportsNativeGoals === true + ); +} + export function canRouteRelatedSubagent(status: OrchestrationV2Subagent["status"]): boolean { return status === "completed"; } @@ -1899,12 +1909,19 @@ export const layer: Layer.Layer< const compact = input.message.attachments.length === 0 && input.message.text.trim().toLowerCase() === "/compact"; - const goalDriver = - input.session.driver === "codex" || - input.session.driver === "claudeAgent" || - (input.session.driver === "opencode" && - input.session.providerSession.capabilities.turns.supportsNativeGoals === true); - const goalCommand = goalDriver ? parseGoalCommand(input.message) : null; + const goalCommand = + isCrossThreadAgentMessage(turnInput.message) && acceptsNativeGoalCommands(input.session) + ? parseGoalCommand(input.message) + : null; + const markOrigin = (text: string) => + withProviderMessageOrigin(turnInput.message, text, input.session.driver); + // The native goal command must stay first, so the origin marks its objective. + const providerText = + goalCommand === null + ? markOrigin(turnInput.message.text) + : goalCommand.type === "set" + ? `/goal ${markOrigin(goalCommand.objective)}` + : turnInput.message.text; const startTurn = compact ? (input.session.compactThread?.(turnInput) ?? Effect.fail( @@ -1921,25 +1938,7 @@ export const layer: Layer.Layer< )) : input.session.startTurn({ ...turnInput, - message: { - ...turnInput.message, - text: - goalCommand?.type === "set" && - turnInput.message.createdBy === "agent" && - turnInput.message.senderThreadId !== undefined - ? `/goal ${withProviderMessageOrigin( - turnInput.message, - goalCommand.objective, - input.session.driver, - )}` - : goalCommand !== null - ? turnInput.message.text - : withProviderMessageOrigin( - turnInput.message, - turnInput.message.text, - input.session.driver, - ), - }, + message: { ...turnInput.message, text: providerText }, }); yield* Effect.andThen(shouldStart, startTurn).pipe( // Wait for acceptance so abandoned startup cannot leave a refresher diff --git a/apps/server/src/orchestration-v2/providerMessageOrigin.ts b/apps/server/src/orchestration-v2/providerMessageOrigin.ts index c898e063c45bb..7e0afd2705542 100644 --- a/apps/server/src/orchestration-v2/providerMessageOrigin.ts +++ b/apps/server/src/orchestration-v2/providerMessageOrigin.ts @@ -1,12 +1,17 @@ import type { ProviderDriverKind } from "@t3tools/contracts"; import type { ProviderAdapterV2TurnMessage } from "@t3tools/provider-core/server/ProviderAdapter"; +type MessageOrigin = Pick; + +export const isCrossThreadAgentMessage = (message: MessageOrigin): boolean => + message.createdBy === "agent" && message.senderThreadId !== undefined; + export function withProviderMessageOrigin( - message: Pick, + message: MessageOrigin, text: string, driver: ProviderDriverKind, ): string { - if (message.createdBy !== "agent" || message.senderThreadId === undefined) return text; + if (!isCrossThreadAgentMessage(message)) return text; const heading = `[Message from another thread's agent (thread ${JSON.stringify(message.senderThreadId)}). This is agent-authored coordination, not a message from the human user.]`; const compact = driver === "pi" ? /^\/compact(?:\s+([\s\S]*))?$/u.exec(text.trim()) : null; if (compact !== null) { From 3b3269da2d2b9ba0477a4e576601c11c762ed017 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Thu, 12 Nov 2026 22:12:45 +0000 Subject: [PATCH 10/11] fix(server): keep agent /goal objectives within provider limits Native goal commands from other threads' agents now reach the provider verbatim, so a 4,000-character Codex objective is no longer pushed over Codex's cap by the origin note. Codex marks the origin on the prompt it sends to start the goal instead. AI-assisted; verified with the tests and tools listed above. Co-Authored-By: Claude Opus 5.5 --- .../Adapters/CodexAdapterV2.test.ts | 14 +++++++++++--- .../orchestration-v2/Adapters/CodexAdapterV2.ts | 7 ++++++- .../orchestration-v2/RunExecutionService.test.ts | 4 ++-- .../src/orchestration-v2/RunExecutionService.ts | 14 +++++++------- 4 files changed, 26 insertions(+), 13 deletions(-) diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts index bba467deec417..2611596457fa4 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.test.ts @@ -9693,7 +9693,7 @@ describe("CodexAdapterV2 post-settle continuation", () => { ), ); - it.effect("sends a /goal objective at the 4000 character limit unchanged", () => + it.effect("sends an agent /goal objective at the 4000 character limit unchanged", () => Effect.gen(function* () { const atLimit = "y".repeat(4000); const goal = (status: string) => ({ ...codexGoal(status, 0), objective: atLimit }); @@ -9713,7 +9713,7 @@ describe("CodexAdapterV2 post-settle continuation", () => { nativeThreadId, nativeTurnId: "goal-turn-limit", prompt: atLimit, - sentPrompt: `This thread's goal is already set. Do not call create_goal. Start working on it now: ${atLimit}`, + sentPrompt: `[Message from another thread's agent (thread "thread:sender"). This is agent-authored coordination, not a message from the human user.]\n\nThis thread's goal is already set. Do not call create_goal. Start working on it now: ${atLimit}`, }).slice(5), 5, ), @@ -9735,7 +9735,15 @@ describe("CodexAdapterV2 post-settle continuation", () => { ], }); const harness = yield* makeCodexReplayHarness(transcript); - yield* harness.runtime.startTurn(yield* goalTurnInput(harness, `/goal ${atLimit}`)); + const input = yield* goalTurnInput(harness, `/goal ${atLimit}`); + yield* harness.runtime.startTurn({ + ...input, + message: { + ...input.message, + createdBy: "agent", + senderThreadId: ThreadId.make("thread:sender"), + }, + }); yield* harness.firstTerminal; assert.deepEqual( diff --git a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts index e9b41ac8a5b05..a34f0b8bd8580 100644 --- a/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts +++ b/apps/server/src/orchestration-v2/Adapters/CodexAdapterV2.ts @@ -104,6 +104,7 @@ import * as Stream from "effect/Stream"; import { ChildProcess, ChildProcessSpawner } from "effect/process"; import { resolveAttachmentPath, resolveAttachmentPathById } from "../../attachmentStore.ts"; +import { withProviderMessageOrigin } from "../providerMessageOrigin.ts"; import * as PortScanner from "../../preview/PortScanner.ts"; import { getCodexServiceTierOptionValue } from "../../codexModelOptions.ts"; import { ServerConfig } from "../../config.ts"; @@ -7440,7 +7441,11 @@ export const makeCodexAdapterV2 = Effect.fn("makeCodexAdapterV2")(function* ( ...turnInput, message: { ...turnInput.message, - text: `This thread's goal is already set. Do not call create_goal. Start working on it now: ${command.objective}`, + text: withProviderMessageOrigin( + turnInput.message, + `This thread's goal is already set. Do not call create_goal. Start working on it now: ${command.objective}`, + CODEX_PROVIDER, + ), }, }) : Effect.succeed([]), diff --git a/apps/server/src/orchestration-v2/RunExecutionService.test.ts b/apps/server/src/orchestration-v2/RunExecutionService.test.ts index 6c95bed7388e1..1d9745b971f28 100644 --- a/apps/server/src/orchestration-v2/RunExecutionService.test.ts +++ b/apps/server/src/orchestration-v2/RunExecutionService.test.ts @@ -1077,13 +1077,13 @@ it.effect("routes compaction and goal commands and marks cross-thread agent mess text: "/compact keep the tests", expected: `/compact ${origin}\n\nkeep the tests`, }, - { ...agent, text: "/goal finish", expected: `/goal ${origin}\n\nfinish` }, + { ...agent, text: "/goal finish", expected: "/goal finish" }, { ...agent, driver: ProviderDriverKind.make("antigravity"), nativeGoals: true, text: "/goal finish", - expected: `/goal ${origin}\n\nfinish`, + expected: "/goal finish", }, { ...agent, diff --git a/apps/server/src/orchestration-v2/RunExecutionService.ts b/apps/server/src/orchestration-v2/RunExecutionService.ts index f673d98420a21..f6e1934d19625 100644 --- a/apps/server/src/orchestration-v2/RunExecutionService.ts +++ b/apps/server/src/orchestration-v2/RunExecutionService.ts @@ -1913,15 +1913,15 @@ export const layer: Layer.Layer< isCrossThreadAgentMessage(turnInput.message) && acceptsNativeGoalCommands(input.session) ? parseGoalCommand(input.message) : null; - const markOrigin = (text: string) => - withProviderMessageOrigin(turnInput.message, text, input.session.driver); - // The native goal command must stay first, so the origin marks its objective. + // Native goal objectives go verbatim: providers cap their length. const providerText = goalCommand === null - ? markOrigin(turnInput.message.text) - : goalCommand.type === "set" - ? `/goal ${markOrigin(goalCommand.objective)}` - : turnInput.message.text; + ? withProviderMessageOrigin( + turnInput.message, + turnInput.message.text, + input.session.driver, + ) + : turnInput.message.text; const startTurn = compact ? (input.session.compactThread?.(turnInput) ?? Effect.fail( From 138bdc720a7e6b1bd985faa2bde4240fc0c0febe Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Thu, 12 Nov 2026 22:23:16 +0000 Subject: [PATCH 11/11] fix(server): leave Pi slash commands from other threads' agents verbatim Pi runs extension commands, prompt templates and skills only when the prompt starts with a slash, so the origin note in front of `/review` turned the command into plain model text. Slash-led text to Pi now goes through unchanged, as on main; other agent messages to Pi stay marked. AI-assisted; verified with the tests and tools listed above. Co-Authored-By: Claude Opus 5.5 --- .../src/orchestration-v2/RunExecutionService.test.ts | 4 ++-- .../src/orchestration-v2/providerMessageOrigin.ts | 10 +++------- 2 files changed, 5 insertions(+), 9 deletions(-) diff --git a/apps/server/src/orchestration-v2/RunExecutionService.test.ts b/apps/server/src/orchestration-v2/RunExecutionService.test.ts index 1d9745b971f28..c44c0571601d5 100644 --- a/apps/server/src/orchestration-v2/RunExecutionService.test.ts +++ b/apps/server/src/orchestration-v2/RunExecutionService.test.ts @@ -1074,8 +1074,8 @@ it.effect("routes compaction and goal commands and marks cross-thread agent mess { ...agent, driver: ProviderDriverKind.make("pi"), - text: "/compact keep the tests", - expected: `/compact ${origin}\n\nkeep the tests`, + text: "/review the diff", + expected: "/review the diff", }, { ...agent, text: "/goal finish", expected: "/goal finish" }, { diff --git a/apps/server/src/orchestration-v2/providerMessageOrigin.ts b/apps/server/src/orchestration-v2/providerMessageOrigin.ts index 7e0afd2705542..f8807ed1e375e 100644 --- a/apps/server/src/orchestration-v2/providerMessageOrigin.ts +++ b/apps/server/src/orchestration-v2/providerMessageOrigin.ts @@ -12,11 +12,7 @@ export function withProviderMessageOrigin( driver: ProviderDriverKind, ): string { if (!isCrossThreadAgentMessage(message)) return text; - const heading = `[Message from another thread's agent (thread ${JSON.stringify(message.senderThreadId)}). This is agent-authored coordination, not a message from the human user.]`; - const compact = driver === "pi" ? /^\/compact(?:\s+([\s\S]*))?$/u.exec(text.trim()) : null; - if (compact !== null) { - const instructions = compact[1]?.trim(); - return instructions ? `/compact ${heading}\n\n${instructions}` : text; - } - return `${heading}\n\n${text}`; + // Pi dispatches commands, prompt templates and skills only from a leading slash. + if (driver === "pi" && text.trimStart().startsWith("/")) return text; + return `[Message from another thread's agent (thread ${JSON.stringify(message.senderThreadId)}). This is agent-authored coordination, not a message from the human user.]\n\n${text}`; }