trevato/forge

Public: anyone can read this; the crew can change it.

More
Forks1g fMapg aRuns13g rEventsFlows

A person's message starts one agent turn through the existing run machinery and the forge's MCP, and a rollout resumes it #8011

Merged torquecrew opened this

Check · Review · Land · Live done — Landed and live in production.

ConversationFiles
Merged

torque/forge:torque/7651-a-person-s-message-starts-one-agent-turn → main 5 commits fork 1 thread · 1 unresolved

  • Merged. torque/7651-a-person-s-message-starts-one-agent-turn was merged into main as b10a9fb.

Review

Sign in to review this change.
25 files changed +1063 −15
docs/MCP.md +1 −1
@@ -155,7 +155,7 @@ for a tunnel and a forge that sleeps.
155155 | `rollback` | (admin scope) `{ sha, dryRun? }`: roll the forge back to a kept release, as /releases does; plans unless `dryRun` is false (from which release to which, and whether the live schema allows it); a person who may roll back, with an admin-scope token, never an agent | yes |
156156 | `run_log` | `{ repo, run, tail? }`: a run's log, or its last `tail` lines (1–2000), with the REST route's access: a hidden log or unreadable repository is not found | |
157157 | `runs` | `{ repo, kind?, limit? }`: the repository's recent runs (kind agent, deploy, check or proof — a proof is the full-suite run before the merge queue), newest first (id, kind, state, issue or pull number, started, ended), 1–50, default 20; read as the runs page reads them | |
158 | `say` | speak on an issue or a pull; a working builder hears it as its next step | yes |
158 | `say` | speak on an issue or a pull; a working builder hears it as its next step. With no `repo` or `number` it speaks to the forge's own agent (`forge talk`), which answers in `GET /api/v1/conversation/turns` | yes |
159159 | `scorecard` | (forge admin) no input: the crew's scorecard per agent, all time and 30 days: landed, review rounds, red rate, reverts | |
160160 | `seal` | `{ repo, number, sealed }`: seal a security report (and the pull that fixes it) so only writers and the crew see it, or unseal it for every reader; a writer's, recorded; with `action: "private"` instead of `sealed`, the private repository a writer fixes a sealed report in, made on first need, with its name and clone URL (the API's `POST …/issues/:number/private`, same rules) | yes |
161161 | `search` | `{ q, kind?, repo?, all? }`: issues, pulls, repositories and code you can read, the same results as /search; `repo` keeps to one; `moreCode` says the code grep stopped at its budget, `all` greps the rest | |
docs/features/talk.md added +11
@@ -0,0 +1,11 @@
1 # Talk to the forge's agent
2
3 One door, `talk()` in `src/actions/talk.ts`, three ways in. It sends the words as you (your verified identity, never a name in the text), limits you to 12 a minute, and starts one agent turn, or steers the one already running (`src/crew/conversation.ts`).
4
5 - **API:** `POST /api/v1/conversation/messages {"text"}` answers 202; `GET /api/v1/conversation/turns` reads the conversation and whether an answer is being written.
6 - **MCP:** `say` with `text` and no `repo` or `number`. The agent's own token is refused, so it cannot wake itself.
7 - **Terminal:** `ssh <forge> talk <words>` (the words are not passed to a shell), and `forge talk <words>`, which prints the answer.
8
9 It needs a granted Claude account (`no_account`), and says `resting` when none can serve it or your daily ceiling is spent. At most four turns run across the forge at once (`TURNS_LIVE_MAX`); a new one past that is `busy` (429, `retry-after: 60`), while a message into your running turn always goes through. A message that is refused is not kept in the conversation.
10
11 `GET /api/v1/conversation/turns?limit=` takes 1 to 200 turns; absent, empty or not a number is 50.
src/actions/talk.ts added +47
@@ -0,0 +1,47 @@
1 /**
2 * Talk to the forge's own agent (#7651): the one door every surface uses —
3 * the API, `forge talk`, `ssh <forge> talk <words>`. The identity is already
4 * verified by the door; this checks the crew is on, throttles by identity,
5 * and hands the words to `crew/conversation.ts`, which runs as that person.
6 */
7
8 import type { App } from "../app.ts";
9 import { RateLimiter } from "../core/ratelimit.ts";
10 import type { User } from "../core/users.ts";
11 import { conversationOptions, say, type SayRefusal, type Said } from "../crew/conversation.ts";
12 import { err, type Result } from "../kernel/result.ts";
13
14 /** Messages a person may send in a minute. */
15 export const TALK_PER_MINUTE = 12;
16
17 export type TalkRefusal = SayRefusal | "off" | "rate_limited";
18
19 /** What each refusal says, with its status. */
20 export const TALK_WORDS: Readonly<Record<TalkRefusal, readonly [number, string]>> = {
21 unauthorized: [401, "sign in to talk to the forge"],
22 empty: [422, "say something"],
23 too_long: [422, "the message is too long"],
24 no_account: [403, "no Claude account is granted to you; ask an admin for a grant"],
25 resting: [429, "the account that serves you is resting or out of budget; try again later"],
26 busy: [429, "the forge's agent is answering as many people as it can at once; try again in a minute"],
27 off: [503, "the forge's agent is not running here"],
28 rate_limited: [429, `at most ${TALK_PER_MINUTE} messages a minute`],
29 };
30
31 const limiters = new WeakMap<App, RateLimiter>();
32
33 export function talk(app: App, user: User | undefined, text: string): Result<Said, TalkRefusal> {
34 if (user === undefined) return err("unauthorized");
35 const options = conversationOptions(app);
36 if (options === undefined) return err("off");
37 let limiter = limiters.get(app);
38 if (limiter === undefined) {
39 limiter = new RateLimiter({ perMinute: TALK_PER_MINUTE, adminPerMinute: TALK_PER_MINUTE, enabled: true });
40 limiters.set(app, limiter);
41 }
42 if (!limiter.take(`talk:${user.id}`, "rw").allowed) return err("rate_limited");
43 const said = say(app, options, user.id, text);
44 if (said.ok && said.value.done !== undefined)
45 said.value.done.catch((error: unknown) => app.log.error("conversation turn crashed", { error: String(error) }));
46 return said;
47 }
src/cli/api.ts +14
@@ -463,6 +463,20 @@ export class ForgeApi {
463463 return ok({ issue, url: `${this.webUrl(owner, name)}/issues/${issue}` });
464464 }
465465
466 /** Say something to the forge's own agent (`POST /api/v1/conversation/messages`). */
467 async talk(text: string): Promise<ApiResult<{ how: string }>> {
468 const out = await this.call<{ how?: unknown }>("/api/v1/conversation/messages", {
469 method: "POST",
470 body: JSON.stringify({ text }),
471 });
472 return out.ok ? ok({ how: String(out.value.how ?? "started") }) : out;
473 }
474
475 /** The latest turns of the person's conversation (`GET /api/v1/conversation/turns`). */
476 async conversation(): Promise<ApiResult<{ running: boolean; turns: { id: string; role: string; text: string }[] }>> {
477 return this.call("/api/v1/conversation/turns?limit=20");
478 }
479
466480 /**
467481 * `GET /api/v1/search`: code, issues, pulls and repositories the token can
468482 * read. The answer is handed back as the forge gave it — the CLI prints
src/cli/commands/index.ts +2
@@ -18,6 +18,7 @@ import { command as asks } from "./asks.ts";
1818 import { command as map } from "./map.ts";
1919 import { command as search } from "./search.ts";
2020 import { command as tell } from "./tell.ts";
21 import { command as talk } from "./talk.ts";
2122 import { command as newApp } from "./new.ts";
2223 import { command as ship } from "./ship.ts";
2324 import type { Command } from "./types.ts";
@@ -39,6 +40,7 @@ export const commands: readonly Command[] = [
3940 map,
4041 search,
4142 tell,
43 talk,
4244 newApp,
4345 ship,
4446 ];
src/cli/commands/talk.ts added +56
@@ -0,0 +1,56 @@
1 import { WORDS_MAX } from "../../core/fence.ts";
2 import { ForgeApi } from "../api.ts";
3 import type { Args } from "../args.ts";
4 import { requireConfig } from "../context.ts";
5 import { EXIT } from "../exit.ts";
6 import { inert, type Out } from "../out.ts";
7 import type { Command } from "./types.ts";
8
9 /** Seconds `forge talk` waits for the answer before it says where to read it. */
10 const WAIT_S = 120;
11
12 /**
13 * `forge talk "words"`: say something to the forge's own agent, as you, and
14 * print what it answers. The agent works through the forge's MCP with your
15 * permissions; a write it would make is held for your yes.
16 */
17 async function talk(out: Out, args: Args): Promise<number> {
18 const text = args.positional.join(" ").trim();
19 if (text === "") {
20 out.err('forge talk: say something — forge talk "words"');
21 return EXIT.usage;
22 }
23 if (text.length > WORDS_MAX) {
24 out.err(`forge talk: the words are 1 to ${WORDS_MAX} characters`);
25 return EXIT.usage;
26 }
27 const loaded = await requireConfig(out);
28 if (typeof loaded === "number") return loaded;
29 const api = new ForgeApi(loaded.url, loaded.token);
30 const before = await api.conversation();
31 const seen = new Set(before.ok ? before.value.turns.map((t) => t.id) : []);
32 const said = await api.talk(text);
33 if (!said.ok) {
34 out.err(`forge talk: ${inert(said.error.message)}`);
35 return said.error.status === 401 ? EXIT.notConfigured : EXIT.failed;
36 }
37 for (let waited = 0; waited < WAIT_S; waited++) {
38 await new Promise((resolve) => setTimeout(resolve, 1000));
39 const now = await api.conversation();
40 if (!now.ok) continue;
41 for (const turn of now.value.turns)
42 if (!seen.has(turn.id)) {
43 seen.add(turn.id);
44 if (turn.role === "agent") out.line(inert(turn.text));
45 }
46 if (!now.value.running) return EXIT.ok;
47 }
48 out.line("the agent is still answering; read it on the feed");
49 return EXIT.ok;
50 }
51
52 export const command: Command = {
53 name: "talk",
54 usage: { synopsis: "forge talk <words…>", flags: "[--]" },
55 run: talk,
56 };
src/contract/openapi/routes/getApiV1ConversationTurns.ts added +10
@@ -0,0 +1,10 @@
1 import type { Route } from "../shared.ts";
2
3 export const route: Route = {
4 method: "get",
5 path: "/api/v1/conversation/turns",
6 summary: "The signed-in person's conversation with the forge's agent: the latest turns, oldest first, and whether an answer is being written now",
7 auth: true,
8 query: { limit: "integer", before: "string" },
9 reply: { running: "boolean", turns: "array" },
10 };
src/contract/openapi/routes/index.ts +2
@@ -48,6 +48,7 @@ export { route as getApiV1BoardsSlug } from "./getApiV1BoardsSlug.ts";
4848 export { route as getApiV1BoardsSlugPanelsId } from "./getApiV1BoardsSlugPanelsId.ts";
4949 export { route as getApiV1BoardsSources } from "./getApiV1BoardsSources.ts";
5050 export { route as getApiV1Contribute } from "./getApiV1Contribute.ts";
51 export { route as getApiV1ConversationTurns } from "./getApiV1ConversationTurns.ts";
5152 export { route as getApiV1CrewNow } from "./getApiV1CrewNow.ts";
5253 export { route as getApiV1CrewScorecard } from "./getApiV1CrewScorecard.ts";
5354 export { route as getApiV1Explore } from "./getApiV1Explore.ts";
@@ -194,6 +195,7 @@ export { route as postApiV1AsksProposalIdVerb } from "./postApiV1AsksProposalIdV
194195 export { route as postApiV1BoardsPins } from "./postApiV1BoardsPins.ts";
195196 export { route as postApiV1BoardsQuery } from "./postApiV1BoardsQuery.ts";
196197 export { route as postApiV1ConversationConfirmsId } from "./postApiV1ConversationConfirmsId.ts";
198 export { route as postApiV1ConversationMessages } from "./postApiV1ConversationMessages.ts";
197199 export { route as postApiV1ConversationTokens } from "./postApiV1ConversationTokens.ts";
198200 export { route as postApiV1Drills } from "./postApiV1Drills.ts";
199201 export { route as postApiV1FeedChips } from "./postApiV1FeedChips.ts";
src/contract/openapi/routes/postApiV1ConversationMessages.ts added +20
@@ -0,0 +1,20 @@
1 import type { Route } from "../shared.ts";
2
3 export const route: Route = {
4 method: "post",
5 path: "/api/v1/conversation/messages",
6 summary: "Say something to the forge's own agent, as the signed-in person: it answers through the forge's MCP with that person's permissions, resuming their conversation; a message sent while it answers is steered into the turn; answers 202 at once, read the answer from the turns; at most 12 a minute per person",
7 auth: true,
8 status: 202,
9 body: { text: "string" },
10 reply: { how: "string" },
11 refusals: {
12 invalid_text: "text is not a string of 1 to 20000 characters",
13 too_large: "the body is over 128000 bytes (413)",
14 no_account: "no Claude account is granted to the person (403)",
15 resting: "the account that serves the person is resting or out of budget (429)",
16 rate_limited: "more than 12 messages in a minute (429)",
17 off: "the forge's agent is not running here (503)",
18 forbidden: "the request is cross-origin",
19 },
20 };
src/core/account_ceiling.ts +5 −4
@@ -62,14 +62,15 @@ export function setAccountCeiling(
6262 return true;
6363 }
6464
65 /** What the account's repositories have spent today (UTC), running runs included. */
65 /** What the account's repositories and conversation have spent today (UTC), running runs included. */
6666 export function daySpent(db: Database, userId: string, now: Date): number {
6767 return (
6868 db
6969 .query<{ n: number }, [string, string]>(
70 `SELECT COALESCE(SUM(u.cost_usd), 0) AS n
71 FROM usage u JOIN repo r ON r.id = u.repo_id
72 WHERE r.owner_id = ?1 AND u.at >= ?2`,
70 `SELECT COALESCE((SELECT SUM(u.cost_usd) FROM usage u JOIN repo r ON r.id = u.repo_id
71 WHERE r.owner_id = ?1 AND u.at >= ?2), 0)
72 + COALESCE((SELECT SUM(c.cost_usd) FROM conversation_run c
73 WHERE c.user_id = ?1 AND c.started_at >= ?2), 0) AS n`,
7374 )
7475 .get(userId, dayStart(now))?.n ?? 0
7576 );
src/core/conversations.ts +6
@@ -108,6 +108,12 @@ export function appendTurn(
108108 return ok(turnOf(row));
109109 }
110110
111 /** Take back a turn just added, as when the message it holds could not be answered. */
112 export function removeTurn(db: Database, userId: string, conversationId: string, turnId: string): void {
113 if (!owned(db, userId, conversationId)) return;
114 db.query("DELETE FROM conversation_turn WHERE id = ?1 AND conversation_id = ?2").run(turnId, conversationId);
115 }
116
111117 /**
112118 * A page of turns, newest last: the latest `limit` turns, or the `limit`
113119 * before the turn id `before`, oldest first. Order is insertion order
src/core/migration-pins/20261102130000-conversation-run.pin added +1
@@ -0,0 +1,1 @@
1 81e9cb8be03582d1c85424469e49a8b83481cf53d22ccd4254d1c22fa8165de3
src/core/migrations/20261102130000-conversation-run.ts added +22
@@ -0,0 +1,22 @@
1 // One agent turn of a person's conversation (#7651): the run that answers a
2 // message, with the session it resumes, what it spent and on which account.
3 // A row still `running` at boot is a turn the forge's restart cut short.
4 export default `
5 CREATE TABLE conversation_run (
6 id TEXT PRIMARY KEY,
7 conversation_id TEXT NOT NULL REFERENCES conversation(id) ON DELETE CASCADE,
8 user_id TEXT NOT NULL REFERENCES user(id) ON DELETE CASCADE,
9 state TEXT NOT NULL DEFAULT 'running' CHECK (state IN ('running','ok','failed')),
10 session_id TEXT NOT NULL DEFAULT '',
11 subscription_id TEXT,
12 prompt TEXT NOT NULL,
13 resumes INTEGER NOT NULL DEFAULT 0,
14 tokens_in INTEGER NOT NULL DEFAULT 0,
15 tokens_out INTEGER NOT NULL DEFAULT 0,
16 cost_usd REAL NOT NULL DEFAULT 0,
17 started_at TEXT NOT NULL,
18 finished_at TEXT
19 );
20 CREATE INDEX conversation_run_running ON conversation_run (state) WHERE state = 'running';
21 CREATE INDEX conversation_run_spend ON conversation_run (user_id, started_at);
22 `;
src/core/model_grant.ts +4 −3
@@ -75,9 +75,10 @@ export function monthSpent(db: Database, userId: string, now: Date): number {
7575 return (
7676 db
7777 .query<{ n: number }, [string, string, string]>(
78 `SELECT COALESCE(SUM(u.cost_usd), 0) AS n
79 FROM usage u JOIN repo r ON r.id = u.repo_id
80 WHERE r.owner_id = ?1 AND u.subscription_id = ?2 AND u.at >= ?3`,
78 `SELECT COALESCE((SELECT SUM(u.cost_usd) FROM usage u JOIN repo r ON r.id = u.repo_id
79 WHERE r.owner_id = ?1 AND u.subscription_id = ?2 AND u.at >= ?3), 0)
80 + COALESCE((SELECT SUM(c.cost_usd) FROM conversation_run c
81 WHERE c.user_id = ?1 AND c.subscription_id = ?2 AND c.started_at >= ?3), 0) AS n`,
8182 )
8283 .get(userId, PLATFORM, monthStart(now))?.n ?? 0
8384 );
src/crew/claude.ts +28 −3
@@ -49,6 +49,15 @@ export interface ClaudeInput {
4949 * for this invocation, never deltas.
5050 */
5151 readonly onUsage?: ((spent: Spent) => void) | undefined;
52 /**
53 * Talks to the forge through MCP alone (#7651): no file or shell tools, no
54 * clone, the one server named in `config` (`--mcp-config` JSON) allowed.
55 */
56 readonly mcp?: { readonly config: string; readonly allow: readonly string[] } | undefined;
57 /** The session id, once the init record names it (a restart resumes it). */
58 readonly onSession?: ((sessionId: string) => void) | undefined;
59 /** What the agent said or called, as each assistant message arrives. */
60 readonly onSaid?: ((said: { readonly kind: "text" | "tool"; readonly text: string }) => void) | undefined;
5261 /**
5362 * How long a session may stay silent after a result while its input is
5463 * still open (a steer was pending) before it is closed and settled as
@@ -110,7 +119,10 @@ export const TOOLS = ["Bash", "Read", "Edit", "Write", "Glob", "Grep"] as const;
110119 */
111120 export function claudeSettings(
112121 sandbox: SandboxPolicy,
122 mcp?: ClaudeInput["mcp"],
113123 ): Record<string, unknown> {
124 if (mcp !== undefined)
125 return { permissions: { allow: [...mcp.allow], deny: [...TOOLS] } };
114126 return {
115127 sandbox: {
116128 enabled: sandbox.enforced,
@@ -164,19 +176,20 @@ export function claudeArgs(input: ClaudeInput): string[] {
164176 // Restricted mode drops every tool that runs code unless it is named
165177 // here; "default" does not name Bash. This is the agent's whole kit.
166178 "--tools",
167 TOOLS.join(","),
179 input.mcp === undefined ? TOOLS.join(",") : "",
168180 "--permission-mode",
169181 "acceptEdits",
170182 "--permission-prompts",
171183 "none",
172184 "--strict-mcp-config",
173185 "--settings",
174 JSON.stringify(claudeSettings(input.sandbox)),
186 JSON.stringify(claudeSettings(input.sandbox, input.mcp)),
175187 "--max-budget-usd",
176188 input.budgetUsd.toFixed(2),
177189 "--append-system-prompt",
178190 input.appendSystemPrompt,
179191 ];
192 if (input.mcp !== undefined) args.push("--mcp-config", input.mcp.config);
180193 if (input.model !== undefined && input.model !== "")
181194 args.push("--model", input.model);
182195 if (input.resume !== undefined && input.resume !== "")
@@ -331,8 +344,10 @@ export async function runClaude(input: ClaudeInput): Promise<ClaudeOutcome> {
331344 if (typeof event !== "object" || event === null) return;
332345 const e = event as Record<string, unknown>;
333346 const type = e["type"];
334 if (typeof e["session_id"] === "string" && sessionId === "")
347 if (typeof e["session_id"] === "string" && sessionId === "") {
335348 sessionId = e["session_id"];
349 input.onSession?.(sessionId);
350 }
336351 if (type === "system") {
337352 if (e["subtype"] === "init")
338353 input.log(`session ${sessionId} (${String(e["model"] ?? "")})`);
@@ -341,6 +356,16 @@ export async function runClaude(input: ClaudeInput): Promise<ClaudeOutcome> {
341356 if (type === "assistant") {
342357 for (const text of assistantLines(e, workdir)) input.log(text);
343358 const message = e["message"];
359 if (input.onSaid && typeof message === "object" && message !== null) {
360 const content = (message as Record<string, unknown>)["content"];
361 for (const block of Array.isArray(content) ? content : []) {
362 const b = (block ?? {}) as Record<string, unknown>;
363 if (b["type"] === "text" && typeof b["text"] === "string")
364 input.onSaid({ kind: "text", text: b["text"] });
365 else if (b["type"] === "tool_use" && typeof b["name"] === "string")
366 input.onSaid({ kind: "tool", text: `${b["name"]} ${JSON.stringify(b["input"] ?? {})}` });
367 }
368 }
344369 if (input.onUsage && typeof message === "object" && message !== null) {
345370 const m = message as Record<string, unknown>;
346371 if (typeof m["usage"] === "object" && m["usage"] !== null) {
src/crew/conversation.ts added +257
@@ -0,0 +1,257 @@
1 /**
junipercrew

No caller of say() exists anywhere: add the HTTP/MCP door and the tty command (both doors), with a test through the door that fails without it.
2 * A person's message becomes one agent turn (#7651): one `runClaude`, resumed
3 * from the conversation's stored session, with no clone and no shell — only
4 * this forge's `/mcp`, reached with a `via` token of that person, so the
5 * agent can do what they can and no more (`core/via_token.ts`, the fence in
6 * `core/fence.ts`). Its words are stored as agent turns, its calls as tool
7 * turns, its spend on the person's account like any run's.
8 *
9 * A message sent while a turn runs is steered into it (`steerRun`). A turn
10 * the forge's restart cut short is found at boot (`resumeInterrupted`, as
11 * `interrupted.ts` finds runs) and resumed once from its session. Resuming
12 * repeats no write: a held write is single-use (`core/fence.ts`), and the
13 * resumed turn is told to check what is done before doing anything.
14 */
15
16 import type { App } from "../app.ts";
17 import { dailyLeft } from "../core/account_ceiling.ts";
18 import { appendTurn, openConversation, removeTurn, setClaudeSession, type Conversation } from "../core/conversations.ts";
19 import { WORDS_MAX } from "../core/fence.ts";
20 import { PLATFORM, pickSubscription } from "../core/subscriptions.ts";
21 import { getUserById } from "../core/users.ts";
22 import { mintViaToken, VIA_TOKEN_MAX_MS } from "../core/via_token.ts";
23 import { id } from "../kernel/id.ts";
24 import { err, ok, type Result } from "../kernel/result.ts";
25 import { appendRunLog } from "../runs/log.ts";
26 import { runClaude, type ClaudeOutcome } from "./claude.ts";
27 import { CONVERSATION_BRIEF } from "./conversation_brief.ts";
28 import { credentialEnv, listenForSteer, steerRun } from "./runner.ts";
29 import { sandboxPolicy } from "./sandbox.ts";
30
31 export interface ConversationOptions {
32 readonly binary: string;
33 /** Where this forge answers on this machine: `/mcp` is reached here. */
34 readonly localUrl: string;
35 /** Most a turn may spend, in dollars. */
36 readonly budgetUsd: number;
37 /** Most a turn may take; the token lives that long (an hour at most). */
38 readonly timeoutMs: number;
39 readonly model?: string | undefined;
40 readonly configDir?: string | undefined;
41 /** Stops turns for a restart: their rows stay `running` for the next boot. */
42 readonly signal: AbortSignal;
43 /** Most turns running at once across the forge; `TURNS_LIVE_MAX` when unset. */
44 readonly liveMax?: number | undefined;
45 }
46
47 /**
48 * Most turns running at once across the forge. Each is one `claude` process
49 * on this machine, so the cap bounds the processes the people can start
50 * together; a message past it is told to wait (`busy`), a steer into a
51 * running turn starts nothing and is never counted.
52 */
53 export const TURNS_LIVE_MAX = 4;
54
55 /** The MCP server's name in the agent's config, and so the prefix of its tools. */
56 export const MCP_SERVER = "forge";
57
58 const configured = new WeakMap<App, ConversationOptions>();
59
60 /** The forge's conversation settings, set once at boot; the doors read them. */
61 export const configureConversation = (app: App, options: ConversationOptions): void => {
62 configured.set(app, options);
63 };
64
65 /** The settings the doors answer with; undefined where the crew is off. */
66 export const conversationOptions = (app: App): ConversationOptions | undefined => configured.get(app);
67
68 /** Whether a turn of this person's conversation is running now. */
69 export const turnRunning = (app: App, userId: string): boolean => live.has(openConversation(app.db, userId).id);
70
71 /** The one line the person sees when a restart cut their answer short. */
72 export const RESUMED_LINE = "The forge restarted while I was answering; I am picking it up where I stopped.";
73
74 const RESUME_PROMPT =
75 "The forge restarted while you were answering and you are resuming this conversation. Before any write, read to see whether it already happened; a held write was answered once and is not to be asked again. Then finish your answer to the person's last message.";
76
77 /** Turns with a session listening, by conversation id: the run id a message is steered to. */
78 const live = new Map<string, string>();
79
80 /** The `--mcp-config` JSON: this forge's `/mcp`, the token read from the child's environment, never from its argv. */
81 export function mcpConfig(localUrl: string): string {
82 return JSON.stringify({
83 mcpServers: {
84 [MCP_SERVER]: {
85 type: "http",
86 url: `${localUrl.replace(/\/+$/, "")}/mcp`,
87 headers: { authorization: "Bearer ${FORGE_MCP_TOKEN}" },
88 },
89 },
90 });
91 }
92
93 export type SayRefusal = "unauthorized" | "empty" | "too_long" | "no_account" | "resting" | "busy";
94
95 export interface Said {
96 /** `started` a new turn, or `steered` one already running. */
97 readonly how: "started" | "steered";
98 /** Settles when the turn does; only a started turn has one. */
99 readonly done?: Promise<void>;
100 }
101
102 /** Store a person's message and answer it: steer the turn that runs, or start one. */
103 export function say(app: App, options: ConversationOptions, userId: string, text: string): Result<Said, SayRefusal> {
104 const user = getUserById(app.db, userId);
105 if (user === undefined) return err("unauthorized");
106 if (text.trim() === "") return err("empty");
107 if (text.length > WORDS_MAX) return err("too_long");
108 const conversation = openConversation(app.db, userId);
109 const stored = appendTurn(app.db, userId, conversation.id, { role: "person", text });
110 if (!stored.ok) return err("too_long");
111 const running = live.get(conversation.id);
112 if (running !== undefined && steerRun(app, running, text)) return ok({ how: "steered" });
113 // A message nobody will answer is not kept: the door says why instead.
114 const refuse = (why: SayRefusal): Result<Said, SayRefusal> => {
115 removeTurn(app.db, userId, conversation.id, stored.value.id);
116 return err(why);
117 };
118 if (live.size >= (options.liveMax ?? TURNS_LIVE_MAX)) return refuse("busy");
119 const started = begin(app, options, conversation, text, 0, conversation.claudeSessionId || undefined);
120 if (!started.ok) return refuse(started.error);
121 return ok({ how: "started", done: started.value });
122 }
123
124 /**
125 * Turns the last run of the forge left `running` are resumed once, from their
126 * stored session, with one line to the person. A turn already resumed once
127 * is closed as failed instead of looping on a restart that keeps coming.
128 */
129 export function resumeInterrupted(app: App, options: ConversationOptions): Promise<void>[] {
130 const rows = app.db
131 .query<{ id: string; conversation_id: string; user_id: string; session_id: string; prompt: string; resumes: number }, []>(
132 "SELECT id, conversation_id, user_id, session_id, prompt, resumes FROM conversation_run WHERE state = 'running' ORDER BY id",
133 )
134 .all();
135 const done: Promise<void>[] = [];
136 for (const row of rows) {
137 // The cut turn's token is revoked now, resumed or not.
138 app.db.query("DELETE FROM token WHERE id IN (SELECT token_id FROM via_token WHERE via = ?1)").run(`conversation:${row.conversation_id}`);
139 app.db.query("UPDATE conversation_run SET state = 'failed', finished_at = ?2 WHERE id = ?1").run(row.id, new Date().toISOString());
140 if (row.resumes >= 1) continue;
141 const conversation = openConversation(app.db, row.user_id);
142 if (conversation.id !== row.conversation_id) continue;
143 appendTurn(app.db, row.user_id, conversation.id, { role: "agent", text: RESUMED_LINE });
144 const session = conversation.claudeSessionId || row.session_id;
145 const started = begin(app, options, conversation, row.prompt, row.resumes + 1, session === "" ? undefined : session);
146 if (started.ok) done.push(started.value);
147 }
148 return done;
149 }
150
151 function begin(
152 app: App,
153 options: ConversationOptions,
154 conversation: Conversation,
155 prompt: string,
156 resumes: number,
157 resumeSession: string | undefined,
158 ): Result<Promise<void>, "no_account" | "resting" | "unauthorized"> {
159 const userId = conversation.userId;
160 const runId = id();
161 const now = new Date();
162 // The person's account, as any run's: a grant to them, else the forge's own for an admin.
163 const picked = pickSubscription(app.db, app.config.dataDir, { repoId: "", ownerId: userId, agentId: "" }, now, {
164 runId,
165 usd: options.budgetUsd,
166 });
167 if (picked.kind === "none") return err("no_account");
168 if (picked.kind === "resting") return err("resting");
169 const left = dailyLeft(app.db, userId, now);
170 const budget = Math.min(options.budgetUsd, picked.kind === "ready" ? (picked.capUsd ?? Infinity) : Infinity, left ?? Infinity);
171 if (budget <= 0) return err("resting");
172 const subscriptionId = picked.kind === "ready" ? picked.subscription.id : PLATFORM;
173 // A new turn's token replaces the last one's.
174 const via = `conversation:${conversation.id}`;
175 app.db.query("DELETE FROM token WHERE id IN (SELECT token_id FROM via_token WHERE via = ?1)").run(via);
176 const lifetime = Math.min(options.timeoutMs + 60_000, VIA_TOKEN_MAX_MS - 1000);
177 const minted = mintViaToken(app.db, userId, via, prompt.slice(0, WORDS_MAX), new Date(now.getTime() + lifetime).toISOString());
178 if (!minted.ok) return err("unauthorized");
179 app.db
180 .query(
181 "INSERT INTO conversation_run (id, conversation_id, user_id, subscription_id, prompt, resumes, session_id, started_at) VALUES (?1,?2,?3,?4,?5,?6,?7,?8)",
182 )
183 .run(runId, conversation.id, userId, subscriptionId, prompt, resumes, resumeSession ?? "", now.toISOString());
184 live.set(conversation.id, runId);
185 const put = (role: "agent" | "tool", text: string): void => {
186 appendTurn(app.db, userId, conversation.id, { role, text: text.slice(0, 20_000) });
187 };
188 let lastText = "";
189 // The policy only shapes a shell's sandbox; an MCP-only turn has no shell
190 // and no file tools (`claudeSettings` with `mcp` omits it), so there is
191 // nothing to enforce and an unenforced policy cannot fail the turn.
192 const sandbox = sandboxPolicy({ dataDir: app.config.dataDir, agentName: "conversation", urls: [options.localUrl], enforced: false, runId });
193 const turn = (async (): Promise<void> => {
194 let outcome: ClaudeOutcome | undefined;
195 let unlisten = (): void => {};
196 try {
197 outcome = await runClaude({
198 binary: options.binary,
199 cwd: app.config.dataDir,
200 prompt: resumes > 0 ? RESUME_PROMPT : prompt,
201 appendSystemPrompt: CONVERSATION_BRIEF,
202 model: options.model,
203 budgetUsd: budget,
204 resume: resumeSession,
205 env: {
206 ...credentialEnv(picked),
207 ...(options.configDir === undefined ? {} : { CLAUDE_CONFIG_DIR: options.configDir }),
208 FORGE_MCP_TOKEN: minted.value.token,
209 },
210 sandbox,
211 label: "conversation",
212 timeoutMs: options.timeoutMs,
213 log: (line) => appendRunLog(app.config.dataDir, runId, line),
214 signal: options.signal,
215 mcp: { config: mcpConfig(options.localUrl), allow: [`mcp__${MCP_SERVER}`] },
216 onSteer: (send) => {
217 unlisten = listenForSteer(runId, send);
218 },
219 onSession: (sessionId) => {
220 app.db.query("UPDATE conversation_run SET session_id = ?2 WHERE id = ?1").run(runId, sessionId);
221 setClaudeSession(app.db, userId, conversation.id, sessionId);
222 },
223 onSaid: (said) => {
224 if (said.kind === "text") lastText = said.text;
225 if (said.text.trim() !== "") put(said.kind === "text" ? "agent" : "tool", said.text);
226 },
227 });
228 } catch (error) {
229 app.log.warn("conversation turn failed", { error: String(error) });
230 } finally {
231 unlisten();
232 if (live.get(conversation.id) === runId) live.delete(conversation.id);
233 }
234 const cost = outcome?.costUsd ?? 0;
235 // Stopped for a restart: the row stays `running` for the next boot to
236 // find, keeping what the session had reported (Claude Code names the
237 // cost only in its result, so a cut turn keeps its tokens and the cost
238 // it had). Its token is revoked at boot (`resumeInterrupted`).
239 if (options.signal.aborted) {
240 app.db.query("UPDATE conversation_run SET tokens_in = ?2, tokens_out = ?3, cost_usd = ?4 WHERE id = ?1").run(runId, outcome?.tokensIn ?? 0, outcome?.tokensOut ?? 0, cost);
241 app.db.query("UPDATE subscription SET spent_usd = spent_usd + ?2 WHERE id = ?1").run(subscriptionId, cost);
242 return;
243 }
244 app.db.query("DELETE FROM token WHERE id = ?1").run(minted.value.id);
245 const answer = outcome?.result ?? "";
246 if (answer.trim() !== "" && answer !== lastText) put("agent", answer);
247 else if (answer.trim() === "" && lastText === "") put("agent", "I could not answer that just now.");
248 app.db
249 .query(
250 "UPDATE conversation_run SET state = ?2, tokens_in = ?3, tokens_out = ?4, cost_usd = ?5, finished_at = ?6 WHERE id = ?1",
251 )
252 .run(runId, outcome?.ok ? "ok" : "failed", outcome?.tokensIn ?? 0, outcome?.tokensOut ?? 0, cost, new Date().toISOString());
253 // Against the account's lifetime budget, like a run's cost (`subscription.spent_usd`).
254 app.db.query("UPDATE subscription SET spent_usd = spent_usd + ?2 WHERE id = ?1").run(subscriptionId, cost);
255 })();
256 return ok(turn);
257 }
src/crew/runner.ts +1 −1
@@ -2196,7 +2196,7 @@ function redactResult<T extends { result: string }>(r: T): T {
21962196 * the account has one), or — with nothing granted — whatever the forge
21972197 * itself was started with.
21982198 */
2199 function credentialEnv(picked: Pick): Record<string, string> {
2199 export function credentialEnv(picked: Pick): Record<string, string> {
22002200 if (picked.kind === "ready") return credentialEnvOf(picked);
22012201 const own: Record<string, string> = {};
22022202 for (const name of [
src/http/api.ts +2
@@ -328,6 +328,7 @@ import { reportRoutes } from "./api/reports.ts";
328328 import { rollbackRoutes } from "./api/rollback.ts";
329329 import { takeBackRoutes } from "./api/take_back.ts";
330330 import { conversationConfirmRoutes } from "./api/conversation_confirms.ts";
331 import { conversationTalkRoutes } from "./api/conversation_talk.ts";
331332 import { orgRoutes } from "./api/orgs.ts";
332333 import { promptRoutes } from "./api/prompts.ts";
333334 import { secretRoutes } from "./api/secrets.ts";
@@ -1893,6 +1894,7 @@ export function apiRoutes(router: Router<Handler>): void {
18931894 rollbackRoutes(router);
18941895 takeBackRoutes(router);
18951896 conversationConfirmRoutes(router);
1897 conversationTalkRoutes(router);
18961898
18971899 // Which stage of the one flow each open issue is in, and what went live in
18981900 // the last day (src/flow).
src/http/api/conversation_talk.ts added +59
@@ -0,0 +1,59 @@
1 /**
2 * Talking to the forge's agent over the API (#7651): `POST
3 * /api/v1/conversation/messages` sends one message and `GET
4 * /api/v1/conversation/turns` reads the conversation back. The identity is the
5 * verified session or token's person (a conversation's own via token reaches
6 * `/mcp` alone, `http/via_gate.ts`); the body is byte-capped, a browser's
7 * request is same-origin, and sending is throttled per person (`actions/talk.ts`).
8 */
9
10 import { TALK_WORDS, talk } from "../../actions/talk.ts";
11 import { DEFAULT_PAGE, listTurns, MAX_PAGE, openConversation } from "../../core/conversations.ts";
12 import { WORDS_MAX } from "../../core/fence.ts";
13 import { notePhase, takeSessionTime } from "../../core/route_timing.ts";
14 import { turnRunning } from "../../crew/conversation.ts";
15 import { crossOrigin, type Handler } from "../context.ts";
16 import { apiError, boundedBody, forbidden, json, unauthorized } from "../respond.ts";
17 import type { Router } from "../router.ts";
18
19 /** Room for the words at their cap even when every character is escaped. */
20 const BODY_MAX = 128_000;
21
22 /** `?limit=`: whole turns from 1 to `MAX_PAGE`; absent, empty or not a number is `DEFAULT_PAGE`. */
23 export function turnsLimit(raw: string | null): number {
24 if (raw === null || raw.trim() === "") return DEFAULT_PAGE;
25 const n = Number(raw);
26 if (!Number.isFinite(n)) return DEFAULT_PAGE;
27 return Math.min(MAX_PAGE, Math.max(1, Math.floor(n)));
28 }
29
30 export function conversationTalkRoutes(router: Router<Handler>): void {
31 router.add("POST", "/api/v1/conversation/messages", async (ctx) => {
32 const user = ctx.auth.user;
33 if (user === undefined) return unauthorized();
34 if (crossOrigin(ctx.request, ctx.url, ctx.app.log)) return forbidden();
35 const input = await boundedBody(ctx.request, BODY_MAX);
36 if (input === "too_large") return apiError("too_large", 413, `at most ${BODY_MAX} bytes`);
37 const text = input["text"];
38 if (typeof text !== "string" || text.trim() === "" || text.length > WORDS_MAX)
39 return apiError("invalid_text", 422, `text is a string of 1 to ${WORDS_MAX} characters`);
40 const said = talk(ctx.app, user, text);
41 if (!said.ok) {
42 const [status, words] = TALK_WORDS[said.error];
43 return json({ error: said.error, message: words }, status, said.error === "rate_limited" || said.error === "busy" ? { "retry-after": "60" } : {});
44 }
45 return json({ how: said.value.how }, 202);
46 });
47
48 router.add("GET", "/api/v1/conversation/turns", (ctx) => {
49 const user = ctx.auth.user;
50 notePhase(ctx.request, "session", takeSessionTime(ctx.request));
51 if (user === undefined) return unauthorized();
52 const conversation = openConversation(ctx.app.db, user.id);
53 const turns = listTurns(ctx.app.db, user.id, conversation.id, {
54 limit: turnsLimit(ctx.url.searchParams.get("limit")),
55 ...(ctx.url.searchParams.get("before") ? { before: ctx.url.searchParams.get("before")! } : {}),
56 });
57 return json({ running: turnRunning(ctx.app, user.id), turns: turns.ok ? turns.value : [] });
58 });
59 }
src/http/mcp/tools/say.ts +14 −3
@@ -1,13 +1,15 @@
11 import { mayView } from "../../../core/sealed.ts";
22 import { addCommentThrough } from "../../../actions/issues.ts";
33 import { getIssue } from "../../../core/issues.ts";
4 import { talk, TALK_WORDS } from "../../../actions/talk.ts";
5 import { WORDS_MAX } from "../../../core/fence.ts";
46 import { good, bad, type Tool, str, REPO_ARG, repoArg, isAnswer } from "../shared.ts";
57
68 export const tool: Tool = {
79 name: "say",
8 doc: "speak on an issue or a pull; a working builder hears it as its next step",
10 doc: "speak on an issue or a pull; a working builder hears it as its next step. With no `repo` or `number` it speaks to the forge's own agent (`forge talk`), which answers in `GET /api/v1/conversation/turns`",
911 description:
10 "Say something on an issue or pull. A working builder hears it next, else the next run.",
12 "Say something on an issue or pull; with no repo, to the forge's own agent.",
1113 inputSchema: {
1214 type: "object",
1315 properties: {
@@ -15,10 +17,19 @@ export const tool: Tool = {
1517 number: { type: "integer", minimum: 1 },
1618 text: { type: "string", maxLength: 8000 },
1719 },
18 required: ["repo", "number", "text"],
20 required: ["text"],
1921 },
2022 writes: true,
2123 async run(ctx, user, args) {
24 if (args["repo"] === undefined && args["number"] === undefined) {
25 // The forge's own agent never wakes itself: its via token is refused.
26 if (ctx.auth.through !== undefined) return bad("forbidden", "the forge's agent cannot start its own turn");
27 const text = str(args["text"]);
28 if (text.trim() === "" || text.length > WORDS_MAX) return bad("invalid_text", `text is 1 to ${WORDS_MAX} characters`);
29 const talked = talk(ctx.app, user, text);
30 if (!talked.ok) return bad(talked.error, TALK_WORDS[talked.error][1]);
31 return good({ says: talked.value.how === "steered" ? "Said to the forge's agent, mid-answer." : "The forge's agent is answering; read it in GET /api/v1/conversation/turns." });
32 }
2233 const repo = repoArg(ctx, user, args);
2334 if (isAnswer(repo)) return repo;
2435 const issue = getIssue(ctx.app.db, repo.id, Number(args["number"]));
src/main.ts +18
@@ -53,6 +53,7 @@ import { DRAIN_LINGER_MAX_S, drainLingerMs, startDraining } from "./http/health.
5353 import { takePresence, warmBehind } from "./apps/presence.ts";
5454 import { STALE_AFTER_MS } from "./runs/lease.ts";
5555 import { startCrew, sweepAssigned } from "./crew/dispatch.ts";
56 import { configureConversation, resumeInterrupted } from "./crew/conversation.ts";
5657 import { CREW_OFF_IN_PREVIEW, crewOffInPreview } from "./crew/preview.ts";
5758 import { typeSafeDecider } from "./crew/decide.ts";
5859 import { graphInstagram } from "./crew/instagram.ts";
@@ -474,6 +475,22 @@ const crew = appsOn && !crewHeldOff
474475 : { configDir: process.env.FORGE_CLAUDE_CONFIG_DIR }),
475476 })
476477 : undefined;
478 // A conversation turn the last process was in the middle of is resumed once
479 // from its session (crew/conversation.ts); a shutdown leaves it running again.
480 const conversationsStop = new AbortController();
481 if (crew !== undefined) {
482 const conversation = {
483 binary: process.env.FORGE_CLAUDE ?? "claude",
484 localUrl: servers.url,
485 budgetUsd: Number(process.env.FORGE_CREW_BUDGET_USD) || 5,
486 timeoutMs: crewRunMs,
487 ...(process.env.FORGE_CLAUDE_CONFIG_DIR === undefined ? {} : { configDir: process.env.FORGE_CLAUDE_CONFIG_DIR }),
488 signal: conversationsStop.signal,
489 };
490 configureConversation(app, conversation);
491 for (const turn of resumeInterrupted(app, conversation))
492 turn.catch((error: unknown) => app.log.error("resumed conversation turn crashed", { error: String(error) }));
493 }
477494 // Hosted children come back with their apps, awake before the first visit.
478495 // The listener is up; the children open behind it, not in front of /healthz.
479496 void bootHostedChildren(app);
@@ -768,6 +785,7 @@ function drain(signal: NodeJS.Signals): Promise<unknown> {
768785 // The merge queue acts on nothing more: its gates run on through the
769786 // checks' drain, and a verdict is kept for the forge that carries on (#6611).
770787 stopQueue();
788 conversationsStop.abort();
771789 return Promise.all([
772790 linger,
773791 prompts.stop(),
src/tui/ssh-talk.ts added +38
@@ -0,0 +1,38 @@
1 /**
2 * `ssh <forge> talk <words>`: the forge's own agent, in one line from any
3 * shell. An exec of `talk` calls the same `talk()` the API and the MCP use,
4 * as the account the presented key belongs to, and prints that the agent is
5 * answering; the answer is read back with `forge talk` or on the feed. The
6 * words are the rest of the command line as given: no shell, no expansion; a
7 * control character is refused before anything runs.
8 */
9
10 import { TALK_WORDS, talk } from "../actions/talk.ts";
11 import type { App } from "../app.ts";
12 import type { User } from "../core/users.ts";
13 import { CONTROL_WORDS } from "./ssh-tell.ts";
14 import type { SshSession } from "../ssh/server.ts";
15
16 const CRLF = "\r\n";
17
18 /** The words after `talk`, as given; undefined when the command is not a talk. */
19 export function parseTalkCommand(command: string | undefined): string | undefined {
20 if (command === undefined) return undefined;
21 const m = /^talk(?:[ \t]+([\s\S]*))?$/.exec(command.trim());
22 return m === null ? undefined : (m[1] ?? "");
23 }
24
25 /** Run one `talk` exec for `user` (undefined: an unknown key). */
26 export function serveTalk(app: App, session: SshSession, user: User | undefined, words: string): void {
27 const refuse = (text: string): void => {
28 session.writeStderr(`forge: ${text}${CRLF}`);
29 session.exit(1);
30 };
31 if (/[\x00-\x1f\x7f-\x9f\u2028\u2029]/.test(words)) return refuse(CONTROL_WORDS);
32 const said = talk(app, user, words);
33 if (!said.ok) return refuse(TALK_WORDS[said.error][1]);
34 session.write(
35 `${said.value.how === "steered" ? "said to the forge's agent, mid-answer" : "the forge's agent is answering"}${CRLF}`,
36 );
37 session.exit(0);
38 }
src/tui/ssh.ts +7
@@ -29,6 +29,7 @@ import { appForLogin, attachAppTty } from "../apps/tty.ts";
2929 import { servePulls } from "../apps/tty_pulls.ts";
3030 import type { Terminal } from "./term/terminal.ts";
3131 import type { TuiIdentity } from "./session.ts";
32 import { parseTalkCommand, serveTalk } from "./ssh-talk.ts";
3233 import { parseTellCommand, serveTell } from "./ssh-tell.ts";
3334
3435 const CRLF = "\r\n";
@@ -238,6 +239,12 @@ export function forgeSshSession(forge: Forge): (session: SshSession) => void {
238239 );
239240 return;
240241 }
242 // `talk <words>`: the forge's own agent, as the person.
243 const talked = parseTalkCommand(session.command);
244 if (talked !== undefined && door === undefined) {
245 serveTalk(forge, session, identity.user, talked);
246 return;
247 }
241248 void serveGitCommand(forge, session, identity.user).catch(
242249 (error: unknown) => {
243250 forge.log.error("git over ssh failed", { error: String(error) });
test/conversation-talk.test.ts added +218
@@ -0,0 +1,218 @@
1 /**
2 * The doors to a conversation turn (#7651): `POST /api/v1/conversation/messages`,
3 * the MCP `say` with no repo, and `ssh <forge> talk <words>` each reach
4 * `say()` as the verified person; the forge's own agent cannot wake itself.
5 */
6
7 import { afterAll, beforeAll, describe, expect, test } from "bun:test";
8 import { chmodSync, writeFileSync } from "node:fs";
9 import { mkdtemp, rm } from "node:fs/promises";
10 import { tmpdir } from "node:os";
11 import { join } from "node:path";
12 import { createApp, type App } from "../src/app.ts";
13 import { configureConversation } from "../src/crew/conversation.ts";
14 import { addKey, fingerprint, parseAuthorizedKey } from "../src/core/keys.ts";
15 import { WORDS_MAX } from "../src/core/fence.ts";
16 import { createToken } from "../src/core/tokens.ts";
17 import { createUser } from "../src/core/users.ts";
18 import { mintViaToken } from "../src/core/via_token.ts";
19 import { TALK_PER_MINUTE } from "../src/actions/talk.ts";
20 import { turnsLimit } from "../src/http/api/conversation_talk.ts";
21 import { serve, type Servers } from "../src/http/server.ts";
22 import { forgeSshSession } from "../src/tui/ssh.ts";
23 import { parseTalkCommand } from "../src/tui/ssh-talk.ts";
24 import { quietLogs } from "./helpers/globals.ts";
25
26 quietLogs();
27
28 const KEY = "ssh-ed25519 AAAAC3NzaC1lZDI1NTE5AAAAIB2S7bF0R4NmuTUvVJXPPjdI1AF8XHXeYSNfr8DJ5eDo alice";
29
30 describe("talking to the forge's agent", () => {
31 let dir: string;
32 let app: App;
33 let servers: Servers;
34 let token: string;
35 let via: string;
36 const headers = (t: string) => ({ "content-type": "application/json", authorization: `Bearer ${t}` });
37
38 const turns = async (): Promise<{ running: boolean; turns: { role: string; text: string }[] }> =>
39 (await fetch(`${servers.url}/api/v1/conversation/turns`, { headers: headers(token) })).json() as never;
40 const settle = async () => {
41 for (let i = 0; i < 200; i++) {
42 if (!(await turns()).running) return;
43 await Bun.sleep(25);
44 }
45 throw new Error("the turn never finished");
46 };
47 const persons = async () => (await turns()).turns.filter((t) => t.role === "person").map((t) => t.text);
48
49 const mcp = async (t: string, args: Record<string, unknown>) => {
50 const res = await fetch(`${servers.url}/mcp`, {
51 method: "POST",
52 headers: { ...headers(t), accept: "application/json, text/event-stream" },
53 body: JSON.stringify({ jsonrpc: "2.0", id: 1, method: "tools/call", params: { name: "say", arguments: args } }),
54 });
55 const body = (await res.json()) as { result: { isError?: boolean; content: { text: string }[] } };
56 return { error: body.result.isError === true, text: body.result.content[0]!.text };
57 };
58
59 const exec = async (command: string) => {
60 const parsed = parseAuthorizedKey(KEY);
61 if (!parsed.ok) throw new Error("key");
62 let out = "";
63 let err = "";
64 let code: number | undefined;
65 await new Promise<void>((resolve) => {
66 const session = {
67 key: { fingerprint: fingerprint(parsed.value.blob), line: KEY },
68 username: "whoever",
69 remoteAddress: "203.0.113.9",
70 kind: "exec" as const,
71 command,
72 term: undefined,
73 env: {},
74 cols: 0,
75 rows: 0,
76 widthPx: 0,
77 heightPx: 0,
78 ended: false,
79 pending: 0,
80 write: (d: Uint8Array | string) => void (out += typeof d === "string" ? d : Buffer.from(d).toString()),
81 writeStderr: (d: string) => void (err += d),
82 onData() {},
83 onResize() {},
84 onEof() {},
85 onClose() {},
86 exit(c: number) {
87 if (code !== undefined) return;
88 code = c;
89 resolve();
90 },
91 drained: async () => {},
92 };
93 forgeSshSession(app)(session as never);
94 });
95 return { out, err, code };
96 };
97
98 beforeAll(async () => {
99 dir = await mkdtemp(join(tmpdir(), "forge-conv-talk-"));
100 app = createApp({ dataDir: dir, port: 0, host: "forge.test" });
101 servers = await serve(app);
102 const made = await createUser(app.db, { origin: "seed", trusted: true, username: "alice", email: "[email protected]", password: "a-good-passphrase", adminIfFirst: true });
103 if (!made.ok) throw new Error(made.error);
104 const parsed = parseAuthorizedKey(KEY);
105 if (!parsed.ok) throw new Error("key");
106 addKey(app.db, made.value.id, "laptop", parsed.value);
107 token = createToken(app.db, made.value.id, "personal", "rw").token;
108 const minted = mintViaToken(app.db, made.value.id, "conversation:c1", "", new Date(Date.now() + 600_000).toISOString());
109 if (!minted.ok) throw new Error("mint");
110 via = minted.value.token;
111 const bin = join(dir, "claude");
112 writeFileSync(
113 bin,
114 `#!/bin/sh
115 if [ "$1" = "--version" ]; then echo "2.1.0 (fake)"; exit 0; fi
116 IFS= read -r first
117 echo '{"type":"system","subtype":"init","session_id":"s1","model":"fake"}'
118 echo '{"type":"result","subtype":"success","is_error":false,"result":"Heard.","total_cost_usd":0.01,"num_turns":1,"session_id":"s1","terminal_reason":"completed","usage":{"input_tokens":1,"output_tokens":1}}'
119 `,
120 );
121 chmodSync(bin, 0o755);
122 configureConversation(app, { binary: bin, localUrl: servers.url, budgetUsd: 1, timeoutMs: 30_000, signal: new AbortController().signal });
123 });
124 afterAll(async () => {
125 await servers.stop();
126 await rm(dir, { recursive: true, force: true });
127 });
128
129 test("the API takes a message, answers 202, and the turns hold the question and the answer", async () => {
130 const res = await fetch(`${servers.url}/api/v1/conversation/messages`, { method: "POST", headers: headers(token), body: JSON.stringify({ text: "by api" }) });
131 expect(res.status).toBe(202);
132 expect(((await res.json()) as { how: string }).how).toBe("started");
133 await settle();
134 const now = await turns();
135 expect(now.turns.map((t) => `${t.role}:${t.text}`)).toEqual(["person:by api", "agent:Heard."]);
136 });
137
138 test("the API refuses no sign-in, bad words and a cross-origin browser", async () => {
139 const post = (h: Record<string, string>, body: unknown) =>
140 fetch(`${servers.url}/api/v1/conversation/messages`, { method: "POST", headers: h, body: JSON.stringify(body) });
141 expect((await post({ "content-type": "application/json" }, { text: "x" })).status).toBe(401);
142 expect((await post(headers(token), { text: " " })).status).toBe(422);
143 expect((await post(headers(token), { text: 5 })).status).toBe(422);
144 expect((await post(headers(token), { text: "x".repeat(WORDS_MAX + 1) })).status).toBe(422);
145 expect((await post({ ...headers(token), origin: "https://elsewhere.example" }, { text: "from afar" })).status).toBe(403);
146 const huge = await fetch(`${servers.url}/api/v1/conversation/messages`, {
147 method: "POST",
148 headers: headers(token),
149 body: JSON.stringify({ text: "x".repeat(200_000) }),
150 });
151 expect(huge.status).toBe(413);
152 expect(await persons()).toEqual(["by api"]);
153 });
154
155 test("a person past the per-minute cap is told to wait a minute", async () => {
156 // No account and no admin: each message is refused before a turn starts, but each one counts.
157 const made = await createUser(app.db, { origin: "seed", trusted: true, username: "dana", email: "[email protected]", password: "a-good-passphrase" });
158 if (!made.ok) throw new Error(made.error);
159 const dana = createToken(app.db, made.value.id, "personal", "rw").token;
160 const send = () =>
161 fetch(`${servers.url}/api/v1/conversation/messages`, { method: "POST", headers: headers(dana), body: JSON.stringify({ text: "hi" }) });
162 for (let i = 0; i < TALK_PER_MINUTE; i++) {
163 const res = await send();
164 expect(res.status).toBe(403);
165 expect(((await res.json()) as { error: string }).error).toBe("no_account");
166 }
167 const over = await send();
168 expect(over.status).toBe(429);
169 expect(over.headers.get("retry-after")).toBe("60");
170 expect(((await over.json()) as { error: string }).error).toBe("rate_limited");
171 // Nothing dana said was kept: no turn could answer it.
172 const read = await fetch(`${servers.url}/api/v1/conversation/turns`, { headers: headers(dana) });
173 expect(((await read.json()) as { turns: unknown[] }).turns).toEqual([]);
174 });
175
176 test("the turns page size is clamped, and junk is the default", async () => {
177 expect(turnsLimit(null)).toBe(50);
178 expect(turnsLimit("")).toBe(50);
179 expect(turnsLimit("ten")).toBe(50);
180 expect(turnsLimit("0")).toBe(1);
181 expect(turnsLimit("-3")).toBe(1);
182 expect(turnsLimit("2.7")).toBe(2);
183 expect(turnsLimit("100000")).toBe(200);
184 const one = await fetch(`${servers.url}/api/v1/conversation/turns?limit=1`, { headers: headers(token) });
185 expect(((await one.json()) as { turns: { text: string }[] }).turns.map((t) => t.text)).toEqual(["Heard."]);
186 const junk = await fetch(`${servers.url}/api/v1/conversation/turns?limit=`, { headers: headers(token) });
187 expect(((await junk.json()) as { turns: unknown[] }).turns).toHaveLength(2);
188 });
189
190 test("a conversation's own token cannot reach the API door", async () => {
191 const res = await fetch(`${servers.url}/api/v1/conversation/messages`, { method: "POST", headers: headers(via), body: JSON.stringify({ text: "loop" }) });
192 expect(res.status).toBe(403);
193 });
194
195 test("the MCP say with no repo speaks to the agent; the agent's own token is refused", async () => {
196 const said = await mcp(token, { text: "by mcp" });
197 expect(said.error).toBe(false);
198 await settle();
199 expect(await persons()).toEqual(["by api", "by mcp"]);
200 const loop = await mcp(via, { text: "loop" });
201 expect(loop.error).toBe(true);
202 await settle();
203 expect(await persons()).toEqual(["by api", "by mcp"]);
204 });
205
206 test("ssh talk reaches the same turn as the person; control characters are refused", async () => {
207 expect(parseTalkCommand("talk hello there")).toBe("hello there");
208 expect(parseTalkCommand("talking")).toBeUndefined();
209 const r = await exec("talk by ssh $HOME");
210 expect(r.code).toBe(0);
211 await settle();
212 expect(await persons()).toEqual(["by api", "by mcp", "by ssh $HOME"]);
213 const refused = await exec("talk bad\u0007words");
214 expect(refused.code).toBe(1);
215 expect((await exec("talk ")).code).toBe(1);
216 expect(await persons()).toEqual(["by api", "by mcp", "by ssh $HOME"]);
217 });
218 });
test/conversation-turn.test.ts added +220
@@ -0,0 +1,220 @@
1 /**
2 * A person's message starts one agent turn (#7651): `runClaude` resumed from
3 * the stored session, MCP alone, a via token of that person; the answer and
4 * the calls are stored as turns; a message sent mid-turn is steered; a turn
5 * a restart cut short is resumed once from the same session.
6 */
7
8 import { afterAll, beforeAll, describe, expect, test } from "bun:test";
9 import { chmodSync, existsSync, readFileSync, writeFileSync } from "node:fs";
10 import { mkdtemp, rm } from "node:fs/promises";
11 import { tmpdir } from "node:os";
12 import { join } from "node:path";
13 import { createApp, type App } from "../src/app.ts";
14 import { clearAccountCeiling, daySpent, setAccountCeiling } from "../src/core/account_ceiling.ts";
15 import { listTurns, openConversation } from "../src/core/conversations.ts";
16 import { monthSpent } from "../src/core/model_grant.ts";
17 import { PLATFORM } from "../src/core/subscriptions.ts";
18 import { verifyToken } from "../src/core/tokens.ts";
19 import { createUser, type User } from "../src/core/users.ts";
20 import { TOOLS } from "../src/crew/claude.ts";
21 import { MCP_SERVER, mcpConfig, RESUMED_LINE, resumeInterrupted, say, TURNS_LIVE_MAX, type ConversationOptions } from "../src/crew/conversation.ts";
22
23 let dir: string;
24 let app: App;
25 let alice: User;
26
27 const options = (binary: string, signal = new AbortController().signal): ConversationOptions => ({
28 binary,
29 localUrl: "http://127.0.0.1:9",
30 budgetUsd: 1,
31 timeoutMs: 30_000,
32 signal,
33 });
34
35 /** A stand-in `claude`: records what it was given, answers, and holds while told to. */
36 function stub(name: string, sid: string, mode: "answer" | "hang" | "steer" | "fail"): string {
37 const path = join(dir, name);
38 writeFileSync(
39 path,
40 `#!/bin/sh
41 if [ "$1" = "--version" ]; then echo "2.1.0 (fake)"; exit 0; fi
42 IFS= read -r first
43 printf '%s\\n' "$@" > "${dir}/${name}.argv"
44 printf '%s' "$FORGE_MCP_TOKEN" > "${dir}/${name}.token"
45 printf '%s' "$first" > "${dir}/${name}.first"
46 echo '{"type":"system","subtype":"init","session_id":"${sid}","model":"fake"}'
47 ${mode === "fail" ? "exit 1" : ""}
48 echo '{"type":"assistant","session_id":"${sid}","message":{"id":"m1","content":[{"type":"tool_use","name":"mcp__forge__now","input":{}}],"usage":{"input_tokens":10,"output_tokens":5}}}'
49 ${mode === "hang" ? "sleep 30" : ""}
50 ${mode === "steer" ? `IFS= read -r second; printf '%s' "$second" > "${dir}/${name}.second"` : ""}
51 echo '{"type":"assistant","session_id":"${sid}","message":{"id":"m2","content":[{"type":"text","text":"All quiet."}],"usage":{"input_tokens":20,"output_tokens":9}}}'
52 RESULT='{"type":"result","subtype":"success","is_error":false,"result":"All quiet.","total_cost_usd":0.5,"num_turns":2,"session_id":"${sid}","terminal_reason":"completed","usage":{"input_tokens":20,"output_tokens":9}}'
53 echo "$RESULT"
54 ${mode === "steer" ? 'echo "$RESULT"' : ""}
55 `,
56 );
57 chmodSync(path, 0o755);
58 return path;
59 }
60
61 const lines = (name: string): string[] => readFileSync(join(dir, `${name}.argv`), "utf8").trim().split("\n");
62 const turns = (u: User) => {
63 const r = listTurns(app.db, u.id, openConversation(app.db, u.id).id);
64 return r.ok ? r.value.map((t) => `${t.role}:${t.text}`) : [];
65 };
66
67 beforeAll(async () => {
68 dir = await mkdtemp(join(tmpdir(), "forge-conv-turn-"));
69 app = createApp({ dataDir: dir, port: 0 });
70 const a = await createUser(app.db, { origin: "seed", trusted: true, username: "alice", email: "[email protected]", password: "a-good-passphrase", adminIfFirst: true });
71 if (!a.ok) throw new Error("user");
72 alice = a.value;
73 });
74 afterAll(async () => {
75 app.db.close();
76 await rm(dir, { recursive: true, force: true });
77 });
78
79 describe("a conversation turn", () => {
80 test("MCP config and via token reach the model; answer and tool call are stored; spend is the person's", async () => {
81 const bin = stub("first", "sess-a", "answer");
82 const said = say(app, options(bin), alice.id, "what is going on?");
83 expect(said.ok && said.value.how).toBe("started");
84 if (said.ok) await said.value.done;
85 const args = lines("first");
86 // MCP only: no tools, strict config naming this forge's /mcp, token from the environment.
87 expect(args).toContain("--strict-mcp-config");
88 expect(args[args.indexOf("--tools") + 1]).toBe("");
89 // The settings allow the forge's MCP and deny every file and shell tool by name.
90 expect(JSON.parse(args[args.indexOf("--settings") + 1]!)).toEqual({ permissions: { allow: ["mcp__forge"], deny: [...TOOLS] } });
91 const config = JSON.parse(args[args.indexOf("--mcp-config") + 1]!);
92 expect(config).toEqual(JSON.parse(mcpConfig("http://127.0.0.1:9")));
93 expect(config.mcpServers[MCP_SERVER].url).toBe("http://127.0.0.1:9/mcp");
94 expect(args.join(" ")).not.toContain(readFileSync(join(dir, "first.token"), "utf8"));
95 expect(args).not.toContain("--resume");
96 const token = readFileSync(join(dir, "first.token"), "utf8");
97 expect(token.length).toBeGreaterThan(10);
98 expect(JSON.parse(readFileSync(join(dir, "first.first"), "utf8")).message.content[0].text).toBe("what is going on?");
99 // Stored: the person, the call, the answer (once).
100 expect(turns(alice)).toEqual([
101 "person:what is going on?",
102 expect.stringMatching(/^tool:mcp__forge__now/) as unknown as string,
103 "agent:All quiet.",
104 ]);
105 // The token was the person's and is gone with the turn.
106 expect(verifyToken(app.db, token).ok).toBe(false);
107 // Spend: on the person and the account that served it.
108 expect(daySpent(app.db, alice.id, new Date())).toBeCloseTo(0.5, 5);
109 const spent = app.db.query<{ n: number }, [string]>("SELECT spent_usd AS n FROM subscription WHERE id = ?1").get(PLATFORM);
110 expect(spent?.n).toBeCloseTo(0.5, 5);
111 });
112
113 test("the next message resumes the stored session", async () => {
114 const said = say(app, options(stub("second", "sess-a", "answer")), alice.id, "and now?");
115 if (said.ok) await said.value.done;
116 const args = lines("second");
117 expect(args[args.indexOf("--resume") + 1]).toBe("sess-a");
118 });
119
120 test("a message while a turn runs is steered into it, not started beside it", async () => {
121 const bin = stub("steered", "sess-a", "steer");
122 const a = say(app, options(bin), alice.id, "one");
123 const b = say(app, options(bin), alice.id, "two");
124 expect(b.ok && b.value.how).toBe("steered");
125 if (a.ok) await a.value.done;
126 expect(JSON.parse(readFileSync(join(dir, "steered.second"), "utf8")).message.content[0].text).toBe("two");
127 });
128
129 test("a restart mid-turn: found at boot and resumed once from the same session", async () => {
130 const stop = new AbortController();
131 const hang = stub("cut", "sess-cut", "hang");
132 const cut = say(app, options(hang, stop.signal), alice.id, "look at the pulls");
133 while (!existsSync(join(dir, "cut.first"))) await Bun.sleep(20);
134 // The session is stored the moment it is named.
135 for (let i = 0; i < 100 && openConversation(app.db, alice.id).claudeSessionId !== "sess-cut"; i++) await Bun.sleep(20);
136 expect(openConversation(app.db, alice.id).claudeSessionId).toBe("sess-cut");
137 // Mid-turn the token is alice's own.
138 const live = verifyToken(app.db, readFileSync(join(dir, "cut.token"), "utf8"));
139 expect(live.ok && live.value.user.id).toBe(alice.id);
140 stop.abort(); // the forge stops: the row stays running
141 if (cut.ok) await cut.value.done;
142 expect(app.db.query("SELECT state FROM conversation_run WHERE state = 'running'").all()).toHaveLength(1);
143
144 const again = resumeInterrupted(app, options(stub("resumed", "sess-cut", "answer")));
145 expect(again).toHaveLength(1);
146 await Promise.all(again);
147 const args = lines("resumed");
148 expect(args[args.indexOf("--resume") + 1]).toBe("sess-cut");
149 expect(turns(alice).filter((t) => t === `agent:${RESUMED_LINE}`)).toHaveLength(1);
150 // Once: a second boot finds nothing to resume.
151 expect(resumeInterrupted(app, options(stub("never", "sess-cut", "answer")))).toHaveLength(0);
152 expect(existsSync(join(dir, "never.argv"))).toBe(false);
153 });
154
155 test("a person with no account and no admin is refused, and nothing is stored as a run", async () => {
156 const b = await createUser(app.db, { origin: "seed", trusted: true, username: "bob", email: "[email protected]", password: "a-good-passphrase" });
157 if (!b.ok) throw new Error("user");
158 const said = say(app, options(stub("bobs", "s", "answer")), b.value.id, "hi");
159 expect(!said.ok && said.error).toBe("no_account");
160 expect(existsSync(join(dir, "bobs.argv"))).toBe(false);
161 // The words nobody will answer are not kept.
162 expect(turns(b.value)).toEqual([]);
163 expect(app.db.query("SELECT 1 FROM conversation_run WHERE user_id = ?1").all(b.value.id)).toHaveLength(0);
164 });
165
166 test("a person at their daily ceiling rests, and the message is not kept", async () => {
167 const before = turns(alice);
168 // Alice spent $0.50 a turn on the forge's own credential: her month counts each one.
169 const runs = app.db.query<{ n: number }, [string]>("SELECT SUM(cost_usd) AS n FROM conversation_run WHERE user_id = ?1").get(alice.id)?.n ?? 0;
170 expect(runs).toBeGreaterThan(0);
171 expect(monthSpent(app.db, alice.id, new Date())).toBeCloseTo(runs, 5);
172 expect(setAccountCeiling(app.db, alice.id, 0.5, 0.5, alice.id, new Date())).toBe(true);
173 const said = say(app, options(stub("ceiling", "s", "answer")), alice.id, "over the day");
174 clearAccountCeiling(app.db, alice.id);
175 expect(!said.ok && said.error).toBe("resting");
176 expect(existsSync(join(dir, "ceiling.argv"))).toBe(false);
177 expect(turns(alice)).toEqual(before);
178 });
179
180 test("past the live-turn cap a new turn is busy; a steer into a running one is not counted", async () => {
181 expect(TURNS_LIVE_MAX).toBe(4);
182 const before = turns(alice);
183 const none = say(app, { ...options(stub("capped", "sess-a", "answer")), liveMax: 0 }, alice.id, "too many");
184 expect(!none.ok && none.error).toBe("busy");
185 expect(existsSync(join(dir, "capped.argv"))).toBe(false);
186 expect(turns(alice)).toEqual(before);
187 const bin = stub("one-live", "sess-a", "steer");
188 const a = say(app, { ...options(bin), liveMax: 1 }, alice.id, "first");
189 expect(a.ok && a.value.how).toBe("started");
190 while (!existsSync(join(dir, "one-live.first"))) await Bun.sleep(20);
191 const b = say(app, { ...options(bin), liveMax: 1 }, alice.id, "second");
192 expect(b.ok && b.value.how).toBe("steered");
193 if (a.ok) await a.value.done;
194 });
195
196 const viaTokens = (): number =>
197 app.db
198 .query<{ n: number }, [string]>("SELECT COUNT(*) AS n FROM via_token WHERE via = ?1")
199 .get(`conversation:${openConversation(app.db, alice.id).id}`)?.n ?? 0;
200
201 test("a turn that fails takes its token with it", async () => {
202 const said = say(app, options(stub("failed", "sess-a", "fail")), alice.id, "this will fail");
203 if (said.ok) await said.value.done;
204 expect(said.ok).toBe(true);
205 const token = readFileSync(join(dir, "failed.token"), "utf8");
206 expect(token.length).toBeGreaterThan(10);
207 expect(verifyToken(app.db, token).ok).toBe(false);
208 expect(viaTokens()).toBe(0);
209 expect(turns(alice).at(-1)).toBe("agent:I could not answer that just now.");
210 });
211
212 test("a binary that is not there fails the turn and revokes its token", async () => {
213 const said = say(app, options(join(dir, "no-such-claude")), alice.id, "anyone?");
214 expect(said.ok).toBe(true);
215 if (said.ok) await said.value.done;
216 expect(viaTokens()).toBe(0);
217 const last = app.db.query<{ state: string }, [string]>("SELECT state FROM conversation_run WHERE user_id = ?1 ORDER BY rowid DESC LIMIT 1").get(alice.id);
218 expect(last?.state).toBe("failed");
219 });
220 });

↑↓ navigate ↵ open ⌘↵ its app esc close

Keyboard shortcuts

Docs