Skip to content

feat(server): send a message to a thread when it settles #2

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
bogorman wants to merge 1 commit into
base: main
Choose a base branch
from
+396 −6
Open

feat(server): send a message to a thread when it settles #2

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
151 changes: 151 additions & 0 deletions apps/server/src/orchestration-v2/ThreadSettlementService.test.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,7 +6,9 @@ import {
EventId,
ProviderInstanceId,
ProviderSessionId,
RunId,
ThreadId,
type MessageId,
type OrchestrationProjectShell,
type OrchestrationV2AppThread,
type OrchestrationV2Command,
Expand Down Expand Up @@ -527,6 +529,8 @@ const makeHarness = Effect.fn("makeThreadSettlementHarness")(function* (options:
const settingsChanges = yield* PubSub.unbounded<ContractServerSettings>();
const mergedPullRequests = yield* PubSub.unbounded<PullRequestService.PullRequestMergeEvent>();
const commands = yield* Ref.make<ReadonlyArray<AutoSettleCommand>>([]);
/** Commands the settle prompt sends: the prompt, an interrupt, the settle after it. */
const promptCommands = yield* Queue.unbounded<OrchestrationV2Command>();
const branchCalls = yield* Ref.make<
ReadonlyArray<{ readonly cwd: string; readonly branch: string }>
>([]);
Expand Down Expand Up @@ -578,6 +582,15 @@ const makeHarness = Effect.fn("makeThreadSettlementHarness")(function* (options:
});

const dispatch: Orchestrator.OrchestratorV2Shape["dispatch"] = (command) => {
if (
command.type === "message.dispatch" ||
command.type === "thread.settle" ||
command.type === "run.interrupt"
) {
return Queue.offer(promptCommands, command).pipe(
Effect.as({ sequence: 1, storedEvents: [] }),
);
}
if (command.type !== "thread.auto-settle") {
return Effect.die(new Error(`Unexpected command: ${command.type}`));
}
Expand Down Expand Up @@ -669,6 +682,7 @@ const makeHarness = Effect.fn("makeThreadSettlementHarness")(function* (options:
invalidatedCwds,
closedIdle,
scriptRuns,
promptCommands,
publishEvent: (event: OrchestrationV2DomainEvent) => PubSub.publish(domainEvents, event),
updateSettings,
publishMerge: PubSub.publish(mergedPullRequests, {
Expand Down Expand Up @@ -1236,6 +1250,143 @@ describe("ThreadSettlementServiceV2 terminals", () => {
}),
),
);

describe("settle prompt", () => {
const settings = { ...DEFAULT_SERVER_SETTINGS, settlePrompt: "/wrap-up" };
const runEvent = (
thread: OrchestrationV2AppThread,
userMessageId: MessageId,
status: "running" | "completed",
): OrchestrationV2DomainEvent => ({
type: "run.updated",
id: EventId.make(`event:run:${thread.id}:${status}`),
threadId: thread.id,
occurredAt: DateTime.makeUnsafe(NOW),
payload: {
id: RunId.make(`run:${thread.id}`),
threadId: thread.id,
ordinal: 2,
providerInstanceId: ProviderInstanceId.make("codex"),
modelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5" },
providerThreadId: null,
userMessageId,
rootNodeId: null,
activeAttemptId: null,
status,
requestedAt: DateTime.makeUnsafe(NOW),
startedAt: null,
completedAt: null,
checkpointId: null,
contextHandoffId: null,
},
});
const takePrompt = (queue: Queue.Queue<OrchestrationV2Command>) =>
Queue.take(queue).pipe(
Effect.map((command) => {
assert.strictEqual(command.type, "message.dispatch");
return command as Extract<OrchestrationV2Command, { readonly type: "message.dispatch" }>;
}),
);

it.effect("sends the prompt, settles again when its turn ends, then cleans up", () =>
Effect.scoped(
Effect.gen(function* () {
const thread = appThread("wrap-up-thread", "settled");
const fixture = yield* makeHarness({
snapshot: makeSnapshot([]),
settings,
currentThreads: [thread],
});

yield* Effect.gen(function* () {
const service = yield* ThreadSettlementService.ThreadSettlementServiceV2;
yield* startHarness(service, fixture.activation, fixture.snapshotReads);
yield* fixture.publishEvent(settledEvent(thread));
const prompt = yield* takePrompt(fixture.promptCommands);
assert.strictEqual(prompt.threadId, thread.id);
assert.strictEqual(prompt.text, "/wrap-up");
assert.strictEqual(prompt.createdBy, "system");
assert.deepStrictEqual(prompt.dispatchMode, { type: "start_immediately" });

// Its turn ends: the thread settles again. Cleanup waits for that settle.
yield* fixture.publishEvent(runEvent(thread, prompt.messageId, "running"));
yield* fixture.publishEvent(runEvent(thread, prompt.messageId, "completed"));
const settle = yield* Queue.take(fixture.promptCommands);
assert.deepStrictEqual(
{ type: settle.type, threadId: "threadId" in settle ? settle.threadId : null },
{ type: "thread.settle", threadId: thread.id },
);
assert.strictEqual(yield* Queue.size(fixture.closedIdle), 0);

// The settle that follows the prompt cleans up and sends nothing more.
const resettled = {
...thread,
settledAt: DateTime.makeUnsafe(Date.parse(NOW) + 60_000),
};
yield* fixture.publishEvent(settledEvent(resettled));
assert.deepStrictEqual(yield* Queue.take(fixture.closedIdle), { threadId: thread.id });
assert.strictEqual(yield* Queue.size(fixture.promptCommands), 0);
}).pipe(Effect.provide(fixture.layer));
}),
),
);

it.effect("sends the prompt once when a settlement is re-emitted", () =>
Effect.scoped(
Effect.gen(function* () {
const thread = appThread("re-emitted-thread", "settled");
const marker = appThread("marker-thread", "settled");
const fixture = yield* makeHarness({
snapshot: makeSnapshot([]),
settings: { ...DEFAULT_SERVER_SETTINGS },
currentThreads: [thread, marker],
});

yield* Effect.gen(function* () {
const service = yield* ThreadSettlementService.ThreadSettlementServiceV2;
yield* startHarness(service, fixture.activation, fixture.snapshotReads);
yield* fixture.updateSettings({ settlePrompt: "/wrap-up" });
yield* fixture.publishEvent(settledEvent(thread));
yield* takePrompt(fixture.promptCommands);
yield* fixture.publishEvent(settledEvent(thread));
// Events run in order: once the marker is prompted, the re-emitted
// settlement has been handled without a second prompt.
yield* fixture.publishEvent(settledEvent(marker));
const next = yield* takePrompt(fixture.promptCommands);
assert.strictEqual(next.threadId, marker.id);
}).pipe(Effect.provide(fixture.layer));
}),
),
);

it.effect("stops a prompt turn that runs past the timeout, then settles", () =>
Effect.scoped(
Effect.gen(function* () {
const thread = appThread("slow-thread", "settled");
const fixture = yield* makeHarness({
snapshot: makeSnapshot([]),
settings,
currentThreads: [thread],
});

yield* Effect.gen(function* () {
const service = yield* ThreadSettlementService.ThreadSettlementServiceV2;
yield* startHarness(service, fixture.activation, fixture.snapshotReads);
yield* fixture.publishEvent(settledEvent(thread));
const prompt = yield* takePrompt(fixture.promptCommands);
yield* fixture.publishEvent(runEvent(thread, prompt.messageId, "running"));
yield* TestClock.adjust(ThreadSettlementService.SETTLE_PROMPT_TIMEOUT_MS);
const interrupt = yield* Queue.take(fixture.promptCommands);
assert.deepStrictEqual(
{ type: interrupt.type, runId: "runId" in interrupt ? interrupt.runId : null },
{ type: "run.interrupt", runId: RunId.make(`run:${thread.id}`) },
);
assert.strictEqual((yield* Queue.take(fixture.promptCommands)).type, "thread.settle");
}).pipe(Effect.provide(fixture.layer));
}),
),
);
});
});

describe("ThreadSettlementServiceV2 single-thread sweeps", () => {
Expand Down
157 changes: 151 additions & 6 deletions apps/server/src/orchestration-v2/ThreadSettlementService.ts
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,10 @@ import { resolveProjectSettings } from "@t3tools/shared/projectSettings";
import { visibleThreadPullRequests } from "@t3tools/shared/threadPullRequests";
import {
CommandId,
MessageId,
type RunId,
type ThreadId,
type OrchestrationV2AppThread,
type OrchestrationV2DomainEvent,
type OrchestrationV2ThreadShell,
} from "@t3tools/contracts";
Expand Down Expand Up @@ -37,6 +40,15 @@ export interface SettlementPullRequest {

const DAY_MS = 24 * 60 * 60 * 1_000;
export const QUEUED_TURN_START_GRACE_MS = 2 * 60 * 1_000;
/** How long a settle prompt's turn may run before it is stopped and the thread settles. */
export const SETTLE_PROMPT_TIMEOUT_MS = 10 * 60 * 1_000;
const TERMINAL_RUN_STATUSES: ReadonlySet<string> = new Set([
"completed",
"interrupted",
"failed",
"cancelled",
"rolled_back",
]);

function toMillis(value: DateTime.Utc | null | undefined): number | null {
return value == null ? null : DateTime.toEpochMillis(value);
Expand Down Expand Up @@ -556,6 +568,135 @@ export const make = Effect.gen(function* () {
runSweep(null, threadId),
);

// Settle prompt: a thread whose project sets one gets that message as one
// more turn when it settles (sending it re-engages the thread), then settles
// again when the turn ends. "running" waits for that turn; "settling" marks
// the settle this service asked for, so it cleans up instead of prompting.
const settlePrompts = new Map<
ThreadId,
{ readonly messageId: MessageId; phase: "running" | "settling"; runId: RunId | null }
>();
// Settling a settled thread re-emits thread.settled with the same settledAt;
// one prompt per settlement.
const settlePromptedAt = new Map<ThreadId, number>();

const settleAfterPrompt = Effect.fn("ThreadSettlementServiceV2.settleAfterPrompt")(
function* (threadId: ThreadId, messageId: MessageId, timedOut: boolean) {
const pending = settlePrompts.get(threadId);
if (pending === undefined || pending.messageId !== messageId || pending.phase !== "running") {
return;
}
pending.phase = "settling";
const uuid = yield* crypto.randomUUIDv4;
// A turn still running at the timeout is stopped; settling refuses active work.
if (timedOut && pending.runId !== null) {
yield* orchestrator.dispatch({
type: "run.interrupt",
commandId: CommandId.make(`server:settle-prompt-interrupt:${threadId}:${uuid}`),
threadId,
runId: pending.runId,
});
}
yield* orchestrator
.dispatch({
type: "thread.settle",
commandId: CommandId.make(`server:settle-prompt-settle:${threadId}:${uuid}`),
threadId,
})
.pipe(
// Refused when the turn left the thread blocked on the user (a question,
// an approval, a queued message): it stays active for them.
Effect.tapError(() => Effect.sync(() => settlePrompts.delete(threadId))),
);
},
(effect, threadId) =>
effect.pipe(
Effect.catchCause((cause) =>
Cause.hasInterruptsOnly(cause)
? Effect.failCause(cause)
: Effect.logWarning("settling after the settle prompt failed", {
threadId,
cause: Cause.pretty(cause),
}),
),
),
);

/** Sends the project's settle prompt; false when there is none to send. */
const sendSettlePrompt = Effect.fn("ThreadSettlementServiceV2.sendSettlePrompt")(
function* (thread: OrchestrationV2AppThread) {
if (thread.archivedAt !== null) return false;
const settledAtMs = toMillis(thread.settledAt);
if (settledAtMs === null || settlePromptedAt.get(thread.id) === settledAtMs) return false;
const prompt = resolveProjectSettings(yield* settingsService.getSettings, thread.projectId)
.settings.settlePrompt;
if (prompt === "") return false;
settlePromptedAt.set(thread.id, settledAtMs);
const uuid = yield* crypto.randomUUIDv4;
const messageId = MessageId.make(`message:settle-prompt:${uuid}`);
settlePrompts.set(thread.id, { messageId, phase: "running", runId: null });
yield* orchestrator
.dispatch({
type: "message.dispatch",
commandId: CommandId.make(`server:settle-prompt:${thread.id}:${uuid}`),
threadId: thread.id,
messageId,
text: prompt,
attachments: [],
dispatchMode: { type: "start_immediately" },
createdBy: "system",
creationSource: "server",
})
.pipe(Effect.tapError(() => Effect.sync(() => settlePrompts.delete(thread.id))));
yield* Effect.sleep(SETTLE_PROMPT_TIMEOUT_MS).pipe(
Effect.andThen(settleAfterPrompt(thread.id, messageId, true)),
Effect.forkDetach,
);
return true;
},
(effect, thread) =>
effect.pipe(
Effect.catchCause((cause) =>
Cause.hasInterruptsOnly(cause)
? Effect.failCause(cause)
: Effect.logWarning("sending the settle prompt failed", {
threadId: thread.id,
cause: Cause.pretty(cause),
}).pipe(Effect.as(false)),
),
),
);

const onThreadSettled = Effect.fn("ThreadSettlementServiceV2.onThreadSettled")(function* (
thread: OrchestrationV2AppThread,
) {
const pending = settlePrompts.get(thread.id);
if (pending?.phase === "running") return;
if (pending?.phase === "settling") {
// The settle that follows the prompt: the thread is done, clean up as usual.
settlePrompts.delete(thread.id);
} else if (yield* sendSettlePrompt(thread)) {
// Cleanup waits for the settle that follows the prompt's turn.
return;
}
yield* cleanUpSettledThread(thread.id);
});

/** A settle prompt's turn ended (any way): settle the thread again. */
const onRunUpdated = (run: {
readonly id: RunId;
readonly threadId: ThreadId;
readonly userMessageId: MessageId;
readonly status: string;
}) => {
const pending = settlePrompts.get(run.threadId);
if (pending === undefined || pending.messageId !== run.userMessageId) return Effect.void;
pending.runId = run.id;
return TERMINAL_RUN_STATUSES.has(run.status)
? settleAfterPrompt(run.threadId, pending.messageId, false)
: Effect.void;
};

// Settling closes the thread's shells that sit at an idle prompt, so they stop
// holding the worktree. A terminal running a command (a dev server, an
// editor) stays for the user to close. Then the project's settle script runs
Expand Down Expand Up @@ -605,20 +746,24 @@ export const make = Effect.gen(function* () {
const processEvent = (event: OrchestrationV2DomainEvent) => {
switch (event.type) {
case "thread.settled":
return cleanUpSettledThread(event.threadId);
return onThreadSettled(event.payload);
case "thread.pull-request-synced":
case "provider-session.detached":
return worker.enqueue(event.threadId);
case "provider-session.updated":
return event.payload.status !== "starting" && event.payload.status !== "running"
? worker.enqueue(event.threadId)
: Effect.void;
case "run.created":
return onRunUpdated(event.payload);
case "run.updated":
return ["completed", "interrupted", "failed", "cancelled", "rolled_back"].includes(
event.payload.status,
)
? worker.enqueue(event.threadId)
: Effect.void;
return onRunUpdated(event.payload).pipe(
Effect.andThen(
TERMINAL_RUN_STATUSES.has(event.payload.status)
? worker.enqueue(event.threadId)
: Effect.void,
),
);
default:
return Effect.void;
}
Expand Down
Loading