Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
5 changes: 5 additions & 0 deletions apps/server/scripts/acp-mock-agent.ts
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,7 @@ const emitXAiAskUserQuestion = process.env.T3_ACP_EMIT_XAI_ASK_USER_QUESTION ===
const failSetConfigOption = process.env.T3_ACP_FAIL_SET_CONFIG_OPTION === "1";
const exitOnSetConfigOption = process.env.T3_ACP_EXIT_ON_SET_CONFIG_OPTION === "1";
const promptResponseText = process.env.T3_ACP_PROMPT_RESPONSE_TEXT;
const promptDelayMs = Number(process.env.T3_ACP_PROMPT_DELAY_MS ?? "0");
const permissionOptionIds = {
allowOnce: process.env.T3_ACP_ALLOW_ONCE_OPTION_ID ?? "allow-once",
allowAlways: process.env.T3_ACP_ALLOW_ALWAYS_OPTION_ID ?? "allow-always",
Expand Down Expand Up @@ -364,6 +365,10 @@ const program = Effect.gen(function* () {
Effect.gen(function* () {
const requestedSessionId = String(request.sessionId ?? sessionId);

if (Number.isFinite(promptDelayMs) && promptDelayMs > 0) {
yield* Effect.sleep(`${promptDelayMs} millis`);
}

if (emitInterleavedAssistantToolCalls) {
const toolCallId = "tool-call-1";

Expand Down
221 changes: 221 additions & 0 deletions apps/server/src/orchestration/Layers/ProjectionPipeline.test.ts
Original file line number Diff line number Diff line change
Expand Up @@ -1232,6 +1232,227 @@ it.layer(BaseTestLayer)("OrchestrationProjectionPipeline", (it) => {
}),
);

it.effect("keeps the turn running across interim assistant messages until the session ends", () =>
Effect.gen(function* () {
const projectionPipeline = yield* OrchestrationProjectionPipeline;
const eventStore = yield* OrchestrationEventStore;
const sql = yield* SqlClient.SqlClient;
const now = "2026-01-01T00:00:00.000Z";
const threadId = ThreadId.make("thread-turn-lifecycle");
const turnId = TurnId.make("turn-lifecycle-1");

yield* eventStore.append({
type: "thread.created",
eventId: EventId.make("evt-tl1"),
aggregateKind: "thread",
aggregateId: threadId,
occurredAt: now,
commandId: CommandId.make("cmd-tl1"),
causationEventId: null,
correlationId: CorrelationId.make("cmd-tl1"),
metadata: {},
payload: {
threadId,
projectId: ProjectId.make("project-turn-lifecycle"),
title: "Turn lifecycle",
modelSelection: {
instanceId: ProviderInstanceId.make("claude"),
model: "claude-opus",
},
runtimeMode: "full-access",
branch: null,
worktreePath: null,
createdAt: now,
updatedAt: now,
},
});

yield* eventStore.append({
type: "thread.session-set",
eventId: EventId.make("evt-tl2"),
aggregateKind: "thread",
aggregateId: threadId,
occurredAt: "2026-01-01T00:00:01.000Z",
commandId: CommandId.make("cmd-tl2"),
causationEventId: null,
correlationId: CorrelationId.make("cmd-tl2"),
metadata: {},
payload: {
threadId,
session: {
threadId,
status: "running",
providerName: "claude",
runtimeMode: "full-access",
activeTurnId: turnId,
lastError: null,
updatedAt: "2026-01-01T00:00:01.000Z",
},
},
});

// Interim assistant message completes mid-turn (commentary between
// tool calls) — the turn must stay running and unsettled.
yield* eventStore.append({
type: "thread.message-sent",
eventId: EventId.make("evt-tl3"),
aggregateKind: "thread",
aggregateId: threadId,
occurredAt: "2026-01-01T00:00:05.000Z",
commandId: CommandId.make("cmd-tl3"),
causationEventId: null,
correlationId: CorrelationId.make("cmd-tl3"),
metadata: {},
payload: {
threadId,
messageId: MessageId.make("message-tl-interim"),
role: "assistant",
text: "interim commentary",
turnId,
streaming: false,
createdAt: "2026-01-01T00:00:05.000Z",
updatedAt: "2026-01-01T00:00:05.000Z",
},
});

yield* projectionPipeline.bootstrap;

const runningRows = yield* sql<{
readonly state: string;
readonly completedAt: string | null;
}>`
SELECT state, completed_at AS "completedAt"
FROM projection_turns
WHERE thread_id = ${threadId} AND turn_id = ${turnId}
`;
assert.deepEqual(runningRows, [{ state: "running", completedAt: null }]);

// The session leaving "running" is the turn-end signal.
yield* eventStore.append({
type: "thread.session-set",
eventId: EventId.make("evt-tl4"),
aggregateKind: "thread",
aggregateId: threadId,
occurredAt: "2026-01-01T00:01:00.000Z",
commandId: CommandId.make("cmd-tl4"),
causationEventId: null,
correlationId: CorrelationId.make("cmd-tl4"),
metadata: {},
payload: {
threadId,
session: {
threadId,
status: "ready",
providerName: "claude",
runtimeMode: "full-access",
activeTurnId: null,
lastError: null,
updatedAt: "2026-01-01T00:01:00.000Z",
},
},
});

yield* projectionPipeline.bootstrap;

const settledRows = yield* sql<{
readonly state: string;
readonly completedAt: string | null;
}>`
SELECT state, completed_at AS "completedAt"
FROM projection_turns
WHERE thread_id = ${threadId} AND turn_id = ${turnId}
`;
assert.deepEqual(settledRows, [
{ state: "completed", completedAt: "2026-01-01T00:01:00.000Z" },
]);
}),
);

it.effect("settles a superseded running turn when a new turn becomes active", () =>
Effect.gen(function* () {
const projectionPipeline = yield* OrchestrationProjectionPipeline;
const eventStore = yield* OrchestrationEventStore;
const sql = yield* SqlClient.SqlClient;
const now = "2026-01-01T00:00:00.000Z";
const threadId = ThreadId.make("thread-turn-supersede");
const oldTurnId = TurnId.make("turn-superseded");
const newTurnId = TurnId.make("turn-steer");

yield* eventStore.append({
type: "thread.created",
eventId: EventId.make("evt-ts1"),
aggregateKind: "thread",
aggregateId: threadId,
occurredAt: now,
commandId: CommandId.make("cmd-ts1"),
causationEventId: null,
correlationId: CorrelationId.make("cmd-ts1"),
metadata: {},
payload: {
threadId,
projectId: ProjectId.make("project-turn-supersede"),
title: "Turn supersede",
modelSelection: {
instanceId: ProviderInstanceId.make("opencode"),
model: "big-pickle",
},
runtimeMode: "full-access",
branch: null,
worktreePath: null,
createdAt: now,
updatedAt: now,
},
});

const appendRunningSessionSet = (eventId: string, turnId: TurnId, updatedAt: string) =>
eventStore.append({
type: "thread.session-set",
eventId: EventId.make(eventId),
aggregateKind: "thread",
aggregateId: threadId,
occurredAt: updatedAt,
commandId: CommandId.make(`cmd-${eventId}`),
causationEventId: null,
correlationId: CorrelationId.make(`cmd-${eventId}`),
metadata: {},
payload: {
threadId,
session: {
threadId,
status: "running",
providerName: "opencode",
runtimeMode: "full-access",
activeTurnId: turnId,
lastError: null,
updatedAt,
},
},
});

yield* appendRunningSessionSet("evt-ts2", oldTurnId, "2026-01-01T00:00:01.000Z");
// A steer: a new turn becomes active without the provider ever
// completing the previous one.
yield* appendRunningSessionSet("evt-ts3", newTurnId, "2026-01-01T00:00:30.000Z");

yield* projectionPipeline.bootstrap;

const rows = yield* sql<{
readonly turnId: string;
readonly state: string;
readonly completedAt: string | null;
}>`
SELECT turn_id AS "turnId", state, completed_at AS "completedAt"
FROM projection_turns
WHERE thread_id = ${threadId}
ORDER BY requested_at
`;
assert.deepEqual(rows, [
{ turnId: oldTurnId, state: "completed", completedAt: "2026-01-01T00:00:30.000Z" },
{ turnId: newTurnId, state: "running", completedAt: null },
]);
}),
);

it.effect("keeps accumulated assistant text when completion payload text is empty", () =>
Effect.gen(function* () {
const projectionPipeline = yield* OrchestrationProjectionPipeline;
Expand Down
Loading
Loading