From 4d5c19e32b40d1e3e2a4efce66884f16dee53713 Mon Sep 17 00:00:00 2001 From: Sawyer Date: Wed, 4 Nov 2026 00:21:46 -0800 Subject: [PATCH 1/5] fix(hub): give each writer a save tool for only its own document Grok read the writer's four-kind union schema as having no fields and sent empty saves. Each document kind now has its own writer package whose write_artifact lists that kind's fields flat; the hub adds the kind. --- .../src/artifacts/approval-package.test.ts | 7 +- apps/hub/src/artifacts/design-draft.test.ts | 14 +- apps/hub/src/artifacts/tools.test.ts | 167 ++++++++++-------- apps/hub/src/artifacts/tools.ts | 95 +++++++--- apps/hub/src/auth/agent-token-bridge.ts | 2 +- .../src/projects/coordinator-upgrade.test.ts | 29 ++- apps/hub/src/projects/coordinator-upgrade.ts | 14 +- .../hub/src/projects/deploy-presenter.test.ts | 14 +- apps/hub/src/projects/deploy-tools.test.ts | 51 +++--- apps/hub/src/projects/deploy-tools.ts | 25 ++- apps/hub/src/projects/errors.test.ts | 13 +- apps/hub/src/workflows/build-packages.ts | 15 +- apps/hub/src/workflows/director.test.ts | 7 +- .../workflows/entries/writer-approaches.ts | 5 + .../workflows/entries/writer-build-plan.ts | 5 + .../workflows/entries/writer-constraints.ts | 5 + .../entries/writer-product-requirements.ts | 5 + apps/hub/src/workflows/entries/writer.ts | 48 ----- apps/hub/src/workflows/install-for-tenant.ts | 19 +- apps/hub/src/workflows/package-name.ts | 10 ++ apps/hub/src/workflows/packaged.test.ts | 16 +- apps/hub/src/workflows/writer-package.ts | 51 ++++++ packages/schemas/src/artifact.ts | 40 ++--- packages/schemas/src/installed-workflows.ts | 32 +++- test/e2e/oauth-model-login.test.ts | 3 +- workflows/writer/README.md | 13 +- 26 files changed, 451 insertions(+), 254 deletions(-) create mode 100644 apps/hub/src/workflows/entries/writer-approaches.ts create mode 100644 apps/hub/src/workflows/entries/writer-build-plan.ts create mode 100644 apps/hub/src/workflows/entries/writer-constraints.ts create mode 100644 apps/hub/src/workflows/entries/writer-product-requirements.ts delete mode 100644 apps/hub/src/workflows/entries/writer.ts create mode 100644 apps/hub/src/workflows/writer-package.ts diff --git a/apps/hub/src/artifacts/approval-package.test.ts b/apps/hub/src/artifacts/approval-package.test.ts index 83a2f554..ac729769 100644 --- a/apps/hub/src/artifacts/approval-package.test.ts +++ b/apps/hub/src/artifacts/approval-package.test.ts @@ -345,7 +345,12 @@ describe("approval package validation and secret scanning", () => { domain: "example.invalid", role: "coordinator", workflows: { - writer: deployment, + writers: { + constraints: deployment, + approaches: deployment, + product_requirements: deployment, + build_plan: deployment, + }, reviewer: deployment, coder: deployment, designer: deployment, diff --git a/apps/hub/src/artifacts/design-draft.test.ts b/apps/hub/src/artifacts/design-draft.test.ts index 153f6057..3cd4469f 100644 --- a/apps/hub/src/artifacts/design-draft.test.ts +++ b/apps/hub/src/artifacts/design-draft.test.ts @@ -175,7 +175,12 @@ test("file_design_option packs per-screen files and drops the untouched starter domain: "example.invalid", role: "designer", workflows: { - writer: deployment, + writers: { + constraints: deployment, + approaches: deployment, + product_requirements: deployment, + build_plan: deployment, + }, reviewer: deployment, coder: deployment, designer: deployment, @@ -456,7 +461,12 @@ test("designer agent writes per-screen files, is held by checkBeforeDone until c domain: "example.invalid", role: "designer", workflows: { - writer: deployment, + writers: { + constraints: deployment, + approaches: deployment, + product_requirements: deployment, + build_plan: deployment, + }, reviewer: deployment, coder: deployment, designer: deployment, diff --git a/apps/hub/src/artifacts/tools.test.ts b/apps/hub/src/artifacts/tools.test.ts index 36c752e3..dcbf9016 100644 --- a/apps/hub/src/artifacts/tools.test.ts +++ b/apps/hub/src/artifacts/tools.test.ts @@ -1,7 +1,9 @@ import { expect, test } from "bun:test"; +import { type } from "arktype"; import type { RuntimeCapabilities } from "@intx/types/runtime-capabilities"; +import { WriterDocumentKind } from "@corbits-builder/schemas/artifact"; import type { WorkflowRole } from "@corbits-builder/schemas/installed-workflows"; -import { createArtifactTools } from "./tools.ts"; +import { createArtifactTools, createWriterArtifactTools } from "./tools.ts"; const deployment = { source: { @@ -41,7 +43,28 @@ const sampleApproaches = { recommended: "simple", }; -function harness(role: WorkflowRole) { +// The fields each writer's save takes. +const WRITER_FIELDS: Record = { + constraints: ["content", "openQuestions", "title"], + approaches: ["approaches", "content", "openQuestions", "title"], + product_requirements: ["content", "openQuestions", "title"], + build_plan: ["content", "openQuestions", "title"], +}; +const WRITER_KINDS = Object.keys(WRITER_FIELDS).map((kind) => + WriterDocumentKind.assert(kind), +); +// A provider that reads only top-level properties sees nothing to fill in a union and sends {}. +const FlatToolSchema = type({ + type: "'object'", + additionalProperties: "false", + properties: "Record", + "anyOf?": "never", +}); + +function harness( + role: WorkflowRole, + tools: ReturnType, +) { const written: unknown[] = []; const fetch = async (path: string, init: RequestInit) => { if (path === "/api/workflow-project-scope") @@ -51,7 +74,12 @@ function harness(role: WorkflowRole) { domain: "example.invalid", role, workflows: { - writer: deployment, + writers: { + constraints: deployment, + approaches: deployment, + product_requirements: deployment, + build_plan: deployment, + }, reviewer: deployment, coder: deployment, designer: deployment, @@ -83,7 +111,7 @@ function harness(role: WorkflowRole) { const capabilities = { resolve: () => ({ resolve: async () => ({ kind: "http", fetch }) }), } as unknown as RuntimeCapabilities; - const instance = createArtifactTools(role)({ capabilities } as never); + const instance = tools({ capabilities } as never); return { definitions: instance.definitions, written, @@ -96,7 +124,10 @@ function harness(role: WorkflowRole) { } test("coordinator write_artifact exposes only problem_brief without approaches", async () => { - const coordinator = harness("coordinator"); + const coordinator = harness( + "coordinator", + createArtifactTools("coordinator"), + ); const definition = coordinator.definitions.find( (item) => item.name === "write_artifact", ); @@ -139,15 +170,16 @@ test("coordinator write_artifact exposes only problem_brief without approaches", }); test("every role's artifact tool definitions declare top-level type: object", () => { - const roles: WorkflowRole[] = [ - "coordinator", - "writer", - "reviewer", - "coder", - "designer", + const tools = [ + ...(["coordinator", "reviewer", "coder", "designer"] as const).map((role) => + harness(role, createArtifactTools(role)), + ), + ...WRITER_KINDS.map((kind) => + harness("writer", createWriterArtifactTools(kind)), + ), ]; - for (const role of roles) { - for (const definition of harness(role).definitions) { + for (const tool of tools) { + for (const definition of tool.definitions) { expect( (definition.inputSchema as { type?: unknown } | undefined)?.type, ).toBe("object"); @@ -155,80 +187,59 @@ test("every role's artifact tool definitions declare top-level type: object", () } }); -test("writer write_artifact exposes only writer kinds and validates approaches", async () => { - const writer = harness("writer"); - const definition = writer.definitions.find( - (item) => item.name === "write_artifact", - ); - expect( - (definition?.inputSchema as { type?: unknown } | undefined)?.type, - ).toBe("object"); - const branches = ( - definition?.inputSchema as - | { - anyOf?: { - additionalProperties?: boolean; - properties?: { - kind?: { const?: string }; - approaches?: unknown; - }; - }[]; - } - | undefined - )?.anyOf; - const branchFor = (kind: string) => - branches?.find((branch) => branch.properties?.kind?.const === kind); - expect(branches).toHaveLength(4); - const approaches = branchFor("approaches"); - expect(approaches?.additionalProperties).toBe(false); - expect(Object.keys(approaches?.properties ?? {}).sort()).toEqual([ - "approaches", - "content", - "kind", - "openQuestions", - "title", - ]); - for (const kind of ["constraints", "product_requirements", "build_plan"]) { - const branch = branchFor(kind); - expect(branch?.additionalProperties).toBe(false); - expect(Object.keys(branch?.properties ?? {}).sort()).toEqual([ - "content", - "kind", - "openQuestions", - "title", - ]); +test("each writer's write_artifact lists only its own fields, at the top level", () => { + for (const [kind, fields] of Object.entries(WRITER_FIELDS)) { + const tools = harness( + "writer", + createWriterArtifactTools(WriterDocumentKind.assert(kind)), + ); + const [definition] = tools.definitions; + const schema = FlatToolSchema.assert(definition?.inputSchema); + expect(Object.keys(schema.properties).sort()).toEqual(fields); } - expect(branchFor("problem_brief")).toBeUndefined(); +}); - const briefAttempt = await writer.write({ - kind: "problem_brief", - title: "Problem Brief", - content: "# Problem Brief", +test("a writer saves without a kind and files under the kind it was deployed for", async () => { + const constraints = harness( + "writer", + createWriterArtifactTools("constraints"), + ); + const named = await constraints.write({ + kind: "build_plan", + title: "Build Plan", + content: "# Build Plan", + }); + expect(named.isError).toBe(true); + expect(named.content).toStartWith("invalid_arguments:"); + const withApproaches = await constraints.write({ + title: "Constraints", + content: "# Constraints", + approaches: sampleApproaches, }); - expect(briefAttempt.isError).toBe(true); - expect(briefAttempt.content).toStartWith("invalid_arguments:"); + expect(withApproaches.isError).toBe(true); + expect(withApproaches.content).toContain("approaches"); + expect( + (await constraints.write({ title: "Constraints", content: "" })).isError, + ).toBe(true); + expect(constraints.written).toHaveLength(0); - for (const kind of ["constraints", "product_requirements", "build_plan"]) { - const rejected = await writer.write({ - kind, - title: "Doc", - content: "# Doc", - approaches: sampleApproaches, - }); - expect(rejected.isError).toBe(true); - expect(rejected.content).toStartWith("invalid_arguments:"); - expect(rejected.content).toContain("approaches"); - } - expect(writer.written).toHaveLength(0); + const saved = await constraints.write({ + title: "Constraints", + content: "# Constraints", + }); + expect(saved.isError).toBeUndefined(); + expect(constraints.written).toEqual([ + { kind: "constraints", title: "Constraints", content: "# Constraints" }, + ]); - const validApproaches = await writer.write({ - kind: "approaches", + const approaches = harness("writer", createWriterArtifactTools("approaches")); + const valid = await approaches.write({ title: "Approaches", content: "# Approaches", approaches: sampleApproaches, }); - expect(validApproaches.isError).toBeUndefined(); - expect(writer.written).toEqual([ + expect(valid.isError).toBeUndefined(); + expect(approaches.written).toEqual([ { kind: "approaches", title: "Approaches", diff --git a/apps/hub/src/artifacts/tools.ts b/apps/hub/src/artifacts/tools.ts index a821173c..4b57d388 100644 --- a/apps/hub/src/artifacts/tools.ts +++ b/apps/hub/src/artifacts/tools.ts @@ -13,7 +13,9 @@ import { ArtifactRead, ArtifactReceipt, MAX_APPROVAL_PACKAGE_BYTES, - WriterDocumentWrite, + WriterDocumentFields, + type DocumentWrite, + type WriterDocumentKind, } from "@corbits-builder/schemas/artifact"; import type { WorkflowRole } from "@corbits-builder/schemas/installed-workflows"; import { stripApprovalPackageBinaries } from "./approval-package.ts"; @@ -54,16 +56,30 @@ const briefWriteDefinition = { "Write the next version of the problem brief. The hub assigns its id, version and sha256 and returns them. Writing a version is not approval.\nopenQuestions lists each question in the document's Open questions section that the person must answer: a short id, the question as they would read it, whyItBlocks, and suggestedAnswers, your recommendation first. A version written without openQuestions leaves none open.", inputSchema: { ...BriefWrite.toJsonSchema() }, }; -const writerWriteDefinition = { - name: "write_artifact", - description: - "Write the next version of the project document of one kind. The hub assigns its id, version and sha256 and returns them. Writing a version is not approval.\nopenQuestions lists each question in the document's Open questions section that the person must answer: a short id, the question as they would read it, whyItBlocks, and suggestedAnswers, your recommendation first. A version written without openQuestions leaves none open.\nOn the approaches document, approaches carries the approaches it compares, for the person to see side by side: each with a short id, its name, a one-line summary, bestFor, its pros, cons and risks as short phrases, and its effort; recommended names the id of the one you recommend.", - inputSchema: { - type: "object" as const, - ...WriterDocumentWrite.toJsonSchema(), +// Each writer's document by name, and what its write_artifact says about the fields only it has. +const WRITER_DOCUMENTS: Record< + WriterDocumentKind, + { name: string; fieldNotes: string } +> = { + constraints: { name: "constraints", fieldNotes: "" }, + approaches: { + name: "approaches document", + fieldNotes: + "\napproaches carries the approaches the document compares, for the person to see side by side: each with a short id, its name, a one-line summary, bestFor, its pros, cons and risks as short phrases, and its effort; recommended names the id of the one you recommend.", }, + product_requirements: { name: "product requirements", fieldNotes: "" }, + build_plan: { name: "build plan", fieldNotes: "" }, }; +function writerWriteDefinition(kind: WriterDocumentKind) { + const { name, fieldNotes } = WRITER_DOCUMENTS[kind]; + return { + name: "write_artifact", + description: `Write the next version of the ${name}. The hub assigns its id, version and sha256 and returns them. Writing a version is not approval.\nopenQuestions lists each question in the document's Open questions section that the person must answer: a short id, the question as they would read it, whyItBlocks, and suggestedAnswers, your recommendation first. A version written without openQuestions leaves none open.${fieldNotes}`, + inputSchema: { ...WriterDocumentFields[kind].toJsonSchema() }, + }; +} + export function failure(callId: string, error: unknown) { if (error instanceof WorkflowHttpError && error.status) return { @@ -79,24 +95,49 @@ export function failure(callId: string, error: unknown) { return projectScopeFailure(callId, error); } -const DEFINITIONS_BY_ROLE: Record< - WorkflowRole, - readonly (typeof readDefinition)[] +// The tools a role gets, and for a role that writes, the kind it files and the fields it checks. +interface ArtifactToolSet { + definitions: readonly (typeof readDefinition)[]; + save?: { + kind: typeof DocumentWrite.infer.kind; + fields: + | typeof BriefWrite + | (typeof WriterDocumentFields)[WriterDocumentKind]; + }; +} + +// A writer's or reviewer's task carries every document it needs in full; a read tool sends +// writers looking for documents that do not exist. +const TOOLS_BY_ROLE: Record< + Exclude, + ArtifactToolSet > = { - coordinator: [readDefinition, briefWriteDefinition], - writer: [writerWriteDefinition], - reviewer: [readDefinition], - coder: [readDefinition], - designer: [readDefinition], - presenter: [readDefinition], + coordinator: { + definitions: [readDefinition, briefWriteDefinition], + save: { kind: "problem_brief", fields: BriefWrite }, + }, + reviewer: { definitions: [readDefinition] }, + coder: { definitions: [readDefinition] }, + designer: { definitions: [readDefinition] }, + presenter: { definitions: [readDefinition] }, }; -export function createArtifactTools( +export function createArtifactTools(role: Exclude) { + return artifactTools(role, TOOLS_BY_ROLE[role]); +} + +// A writer is deployed for one kind and saves only that kind. +export function createWriterArtifactTools(kind: WriterDocumentKind) { + return artifactTools("writer", { + definitions: [writerWriteDefinition(kind)], + save: { kind, fields: WriterDocumentFields[kind] }, + }); +} + +function artifactTools( role: WorkflowRole, + { definitions, save }: ArtifactToolSet, ): AnnotatedToolFactory { - // A writer's or reviewer's task carries every document it needs in full; a read tool sends - // writers looking for documents that do not exist. - const definitions = DEFINITIONS_BY_ROLE[role]; return defineTool({ id: "@corbits-builder/artifacts/tools", credentials: [{ handle: HUB_CREDENTIAL_HANDLE }], @@ -138,7 +179,7 @@ export function createArtifactTools( maxResponseBytes: MAX_RESULT_BYTES, signal, }); - if (call.name === "write_artifact") { + if (call.name === "write_artifact" && save) { // The coordinator's writers own every kind but the brief, so the documents the person // decides on come from them. const rawKind = (call.arguments as { kind?: unknown } | undefined) @@ -153,16 +194,14 @@ export function createArtifactTools( content: `brief_only: you write the problem brief; deploy_writer has a writer write the ${rawKind}`, isError: true, }; - const input = - role === "coordinator" - ? BriefWrite(call.arguments) - : WriterDocumentWrite(call.arguments); - if (input instanceof type.errors) + const fields = save.fields(call.arguments); + if (fields instanceof type.errors) return { callId: call.id, - content: "invalid_arguments: " + input.summary, + content: "invalid_arguments: " + fields.summary, isError: true, }; + const input = { ...fields, kind: save.kind }; const body = JSON.stringify(input); const earlier = written.get(input.kind); if (earlier?.body === body) diff --git a/apps/hub/src/auth/agent-token-bridge.ts b/apps/hub/src/auth/agent-token-bridge.ts index 90cf38ad..e6af601b 100644 --- a/apps/hub/src/auth/agent-token-bridge.ts +++ b/apps/hub/src/auth/agent-token-bridge.ts @@ -293,7 +293,7 @@ export function createAgentTokenBridge( projectId: binding.projectId, role, workflows: { - writer: installed.writer, + writers: installed.writers, reviewer: installed.reviewer, coder: installsAnyPackage(preferences.value) ? installed.coderAnyPackage diff --git a/apps/hub/src/projects/coordinator-upgrade.test.ts b/apps/hub/src/projects/coordinator-upgrade.test.ts index 41ed89ec..49fc02de 100644 --- a/apps/hub/src/projects/coordinator-upgrade.test.ts +++ b/apps/hub/src/projects/coordinator-upgrade.test.ts @@ -16,7 +16,20 @@ const deployment = (pin: string) => ({ const installed: InstalledWorkflows = { schemaVersion: 1, coordinator: deployment("@corbits-builder/coordinator-native@1.0.0-sha-new"), - writer: deployment("@corbits-builder/writer-native@1.0.0-sha-new"), + writers: { + constraints: deployment( + "@corbits-builder/writer-constraints-native@1.0.0-sha-new", + ), + approaches: deployment( + "@corbits-builder/writer-approaches-native@1.0.0-sha-new", + ), + product_requirements: deployment( + "@corbits-builder/writer-product-requirements-native@1.0.0-sha-new", + ), + build_plan: deployment( + "@corbits-builder/writer-build-plan-native@1.0.0-sha-new", + ), + }, reviewer: deployment("@corbits-builder/reviewer-native@1.0.0-sha-new"), coder: deployment("@corbits-builder/coder-native@1.0.0-sha-new"), coderAnyPackage: deployment("@corbits-builder/coder-native@1.0.0-sha-any"), @@ -54,11 +67,25 @@ test("a run is stale when its frozen pin is not one installed for its role", () role: "designer" as const, bundle: frozen("@corbits-builder/designer-native", "1.0.0-sha-old"), }, + { + runId: "approaches-writer", + role: "writer" as const, + bundle: frozen( + "@corbits-builder/writer-approaches-native", + "1.0.0-sha-new", + ), + }, + { + runId: "single-writer", + role: "writer" as const, + bundle: frozen("@corbits-builder/writer-native", "1.0.0-sha-old"), + }, { runId: "unreadable", role: "writer" as const, bundle: {} }, ]; expect(staleRuns(runs, installed).map((run) => run.runId)).toEqual([ "old-coder", "old-designer", + "single-writer", "unreadable", ]); }); diff --git a/apps/hub/src/projects/coordinator-upgrade.ts b/apps/hub/src/projects/coordinator-upgrade.ts index f2dc0314..0c536f96 100644 --- a/apps/hub/src/projects/coordinator-upgrade.ts +++ b/apps/hub/src/projects/coordinator-upgrade.ts @@ -13,9 +13,10 @@ import { type WorkflowRunReader, } from "@intx/hub-sessions"; import { getLogger } from "@intx/log"; -import type { - InstalledWorkflows, - WorkflowRole, +import { + roleDeployments, + type InstalledWorkflows, + type WorkflowRole, } from "@corbits-builder/schemas/installed-workflows"; import type { ProjectUpdateReceipt } from "@corbits-builder/schemas/project"; import { spawnedWorkflowRole } from "../workflows/spawn-role.ts"; @@ -50,11 +51,10 @@ const pinOf = (bundle: unknown) => { return top && top.name + "@" + top.version; }; -// The pins a role may run on: a coder is either coder package, by the project's package choice. +// The pins a role may run on: a coder is either coder package, by the project's package choice, and a +// writer the package for the kind it was deployed for. const pinsFor = (installed: InstalledWorkflows, role: WorkflowRole) => - role === "coder" - ? [installed.coder.pin, installed.coderAnyPackage.pin] - : [installed[role].pin]; + roleDeployments(installed, role).map((deployment) => deployment.pin); /** The runs whose frozen package is not the one installed for their role, or has no readable pin. */ export function staleRuns( diff --git a/apps/hub/src/projects/deploy-presenter.test.ts b/apps/hub/src/projects/deploy-presenter.test.ts index 32e53e81..e74153c4 100644 --- a/apps/hub/src/projects/deploy-presenter.test.ts +++ b/apps/hub/src/projects/deploy-presenter.test.ts @@ -200,7 +200,12 @@ test("deploy_presenter requires approved documents and pins source refs and file projectId: "project", domain: "example.invalid", workflows: { - writer: deployment, + writers: { + constraints: deployment, + approaches: deployment, + product_requirements: deployment, + build_plan: deployment, + }, reviewer: deployment, coder: deployment, designer: deployment, @@ -281,7 +286,12 @@ test("file_package refuses until check_deck passes on the current deck, then fil domain: "example.invalid", role: "presenter", workflows: { - writer: deployment, + writers: { + constraints: deployment, + approaches: deployment, + product_requirements: deployment, + build_plan: deployment, + }, reviewer: deployment, coder: deployment, designer: deployment, diff --git a/apps/hub/src/projects/deploy-tools.test.ts b/apps/hub/src/projects/deploy-tools.test.ts index 8ef18209..a92540e5 100644 --- a/apps/hub/src/projects/deploy-tools.test.ts +++ b/apps/hub/src/projects/deploy-tools.test.ts @@ -13,6 +13,21 @@ const deployment = { sourceOfferingIds: ["offering"], defaultSourceOfferingId: "offering", }; +const writerDeployment = (name: string) => ({ + ...deployment, + pin: `@corbits-builder/${name}-native@1.0.0`, +}); +const workflows = { + writers: { + constraints: writerDeployment("writer-constraints"), + approaches: writerDeployment("writer-approaches"), + product_requirements: writerDeployment("writer-product-requirements"), + build_plan: writerDeployment("writer-build-plan"), + }, + reviewer: deployment, + coder: deployment, + designer: deployment, +}; type Filed = { version: number; approvedVersion?: number } | undefined; @@ -65,12 +80,7 @@ function coderSpawn(changed: Record) { tenantId: "tenant", projectId: "project", domain: "example.invalid", - workflows: { - writer: deployment, - reviewer: deployment, - coder: deployment, - designer: deployment, - }, + workflows, maxResultBytes: 16384, }); return { @@ -167,12 +177,7 @@ test("design_app starts one designer per concept within the person's limit, each tenantId: "tenant", projectId: "project", domain: "example.invalid", - workflows: { - writer: deployment, - reviewer: deployment, - coder: deployment, - designer: deployment, - }, + workflows, maxResultBytes: 16384, })({ capabilities } as never); const run = (name: string, args: Record) => @@ -220,7 +225,7 @@ function writerTools(preferences: Record = {}) { const fetch = async (path: string, init: RequestInit) => { const body = init.body ? JSON.parse(String(init.body)) : undefined; if (path.endsWith("/workflows/deployments")) { - deployed.push(path); + deployed.push(body.pin); return Response.json({ id: "run-writer", tenantId: "tenant", @@ -274,12 +279,7 @@ function writerTools(preferences: Record = {}) { tenantId: "tenant", projectId: "project", domain: "example.invalid", - workflows: { - writer: deployment, - reviewer: deployment, - coder: deployment, - designer: deployment, - }, + workflows, maxResultBytes: 16384, })({ capabilities } as never); const run = (name: string, args: Record) => @@ -329,7 +329,18 @@ test("one ask approves several documents, and a writer starts once every earlier expect( (await run("deploy_writer", writer("constraints"))).isError, ).toBeUndefined(); - expect(deployed).toHaveLength(2); + expect(deployed).toEqual([ + "@corbits-builder/writer-approaches-native@1.0.0", + "@corbits-builder/writer-constraints-native@1.0.0", + ]); +}); + +test("deploy_writer has no writer for the reviewer's engineering review", async () => { + const { run, deployed } = writerTools({ agent_decides_approvals: "yes" }); + expect( + (await run("deploy_writer", writer("engineering_review"))).content, + ).toStartWith("no_writer_for_kind:"); + expect(deployed).toEqual([]); }); test("with the agent deciding approvals, a filed earlier document is enough", async () => { diff --git a/apps/hub/src/projects/deploy-tools.ts b/apps/hub/src/projects/deploy-tools.ts index e98d6886..fa59f030 100644 --- a/apps/hub/src/projects/deploy-tools.ts +++ b/apps/hub/src/projects/deploy-tools.ts @@ -23,6 +23,7 @@ import { DesignBundle, DesignConcept, MATERIAL_KIND, + WriterDocumentKind, type ArtifactRef, } from "@corbits-builder/schemas/artifact"; import { Preferences } from "@corbits-builder/schemas/preferences"; @@ -236,7 +237,7 @@ export function createDeployTools(options?: { projectId: string; domain: string; workflows: { - writer: NativeSpecialistDeployment; + writers: Record; reviewer: NativeSpecialistDeployment; coder: NativeSpecialistDeployment; designer: NativeSpecialistDeployment; @@ -244,13 +245,18 @@ export function createDeployTools(options?: { }; maxResultBytes: number; }): AnnotatedToolFactory { - if (options) - for (const deployment of Object.values(options.workflows)) { + if (options) { + const { writers, ...specialists } = options.workflows; + for (const deployment of [ + ...Object.values(writers), + ...Object.values(specialists), + ]) { if (!deployment) continue; const checked = NativeSpecialistDeployment.assert(deployment); if (!checked.sourceOfferingIds.includes(checked.defaultSourceOfferingId)) throw new Error("invalid_specialist_default_offering"); } + } const definitions = [ { name: "request_approval", @@ -472,12 +478,12 @@ export function createDeployTools(options?: { // run, so the deploy response is enough to send the assignment. const deploy = async ( role: "writer" | "reviewer" | "coder" | "designer" | "presenter", + workflowSpec: NativeSpecialistDeployment | undefined, preferences: Preferences, correlationId: string, content: string, beforeAssign?: (runId: string) => Promise, ) => { - const workflowSpec = options.workflows[role]; if (!workflowSpec) throw new Refusal( role + "_unavailable: no " + role + " workflow is installed", @@ -533,6 +539,12 @@ export function createDeployTools(options?: { result = { approved: documents }; } else if (call.name === "deploy_writer") { const input = validArguments(DeployWriter, call.arguments); + const writerKind = WriterDocumentKind(input.kind); + if (writerKind instanceof type.errors) + throw new Refusal( + "no_writer_for_kind: review_artifact has the reviewer write the " + + input.kind, + ); const { byKind, approved, preferences } = await listed(); const briefRef = required(byKind, "problem_brief"); const position = DOCUMENT_ORDER.indexOf( @@ -575,6 +587,7 @@ export function createDeployTools(options?: { .map(([, ref]) => ref); result = await deploy( "writer", + options.workflows.writers[writerKind], preferences, "writer-" + input.kind, await assignmentToWriterMessage( @@ -603,6 +616,7 @@ export function createDeployTools(options?: { const target = required(byKind, input.kind); result = await deploy( "reviewer", + options.workflows.reviewer, preferences, "reviewer-" + input.kind, await assignmentToWriterMessage( @@ -647,6 +661,7 @@ export function createDeployTools(options?: { const design = required(byKind, "design"); return deploy( "coder", + options.workflows.coder, preferences, "coder", startHere(inputs(CODER_INPUTS, byKind), design.version) + @@ -724,6 +739,7 @@ export function createDeployTools(options?: { } const started = await deploy( "designer", + options.workflows.designer, preferences, "designer", startHere(inputs(DESIGNER_INPUTS, byKind)) + @@ -815,6 +831,7 @@ export function createDeployTools(options?: { const assignment = buildPackageAssignment(assignmentPayload); result = await deploy( "presenter", + options.workflows.presenter, preferences, "presenter-" + input.audience, presenterAssignmentMessage(assignmentPayload), diff --git a/apps/hub/src/projects/errors.test.ts b/apps/hub/src/projects/errors.test.ts index c32889ac..75d87325 100644 --- a/apps/hub/src/projects/errors.test.ts +++ b/apps/hub/src/projects/errors.test.ts @@ -11,7 +11,7 @@ import { import { ProjectExportMaterial } from "@corbits-builder/schemas/project"; import { mountArtifactRoutes } from "../artifacts/routes.ts"; import { createArtifactService } from "../artifacts/service.ts"; -import { createArtifactTools } from "../artifacts/tools.ts"; +import { createWriterArtifactTools } from "../artifacts/tools.ts"; import { ProjectError, typeIssues } from "./errors.ts"; const deployment = { @@ -63,7 +63,12 @@ test("a rejected write_artifact call reaches the agent with {path, message} issu domain: "example.invalid", role: "writer", workflows: { - writer: deployment, + writers: { + constraints: deployment, + approaches: deployment, + product_requirements: deployment, + build_plan: deployment, + }, reviewer: deployment, coder: deployment, designer: deployment, @@ -79,7 +84,7 @@ test("a rejected write_artifact call reaches the agent with {path, message} issu }, }), ); - const tools = createArtifactTools("writer")({ + const tools = createWriterArtifactTools("approaches")({ capabilities: { resolve: () => ({ resolve: async () => ({ @@ -105,7 +110,6 @@ test("a rejected write_artifact call reaches the agent with {path, message} issu expect( await write({ - kind: "approaches", approaches: { approaches: [approach("a"), approach("a")] }, }), ).toEqual({ @@ -120,7 +124,6 @@ test("a rejected write_artifact call reaches the agent with {path, message} issu }); expect( await write({ - kind: "approaches", approaches: { approaches: [approach("a"), approach("b")], recommended: "c", diff --git a/apps/hub/src/workflows/build-packages.ts b/apps/hub/src/workflows/build-packages.ts index ee0e4e0a..2a9fa933 100644 --- a/apps/hub/src/workflows/build-packages.ts +++ b/apps/hub/src/workflows/build-packages.ts @@ -2,7 +2,7 @@ import { createHash } from "node:crypto"; import { mkdir, mkdtemp, rm, writeFile } from "node:fs/promises"; import { join, resolve } from "node:path"; import { tmpdir } from "node:os"; -import { nativePackageName } from "./package-name.ts"; +import { nativePackageName, WRITER_PACKAGES } from "./package-name.ts"; import type { RoleInference } from "./role-inference.ts"; import type { WorkflowRole } from "@corbits-builder/schemas/installed-workflows"; @@ -15,18 +15,15 @@ function contentVersion(base: string, bundle: string): string { // Each package is an entry under entries/ built on what its role's model binding resolved to. const WORKFLOW_PACKAGES: readonly { name: string; role: WorkflowRole }[] = [ ...( - [ - "coordinator", - "writer", - "reviewer", - "coder", - "designer", - "presenter", - ] as const + ["coordinator", "reviewer", "coder", "designer", "presenter"] as const ).map((role) => ({ name: role, role, })), + ...Object.values(WRITER_PACKAGES).map((name) => ({ + name, + role: "writer" as const, + })), { name: "coder-any-package", role: "coder" }, ]; diff --git a/apps/hub/src/workflows/director.test.ts b/apps/hub/src/workflows/director.test.ts index fee17801..22b72e9b 100644 --- a/apps/hub/src/workflows/director.test.ts +++ b/apps/hub/src/workflows/director.test.ts @@ -443,7 +443,12 @@ test("check_design validates DESIGN.md and .dark in styles.css and auto-files th domain: "example.invalid", role: "designer", workflows: { - writer: deployment, + writers: { + constraints: deployment, + approaches: deployment, + product_requirements: deployment, + build_plan: deployment, + }, reviewer: deployment, coder: deployment, designer: deployment, diff --git a/apps/hub/src/workflows/entries/writer-approaches.ts b/apps/hub/src/workflows/entries/writer-approaches.ts new file mode 100644 index 00000000..04687755 --- /dev/null +++ b/apps/hub/src/workflows/entries/writer-approaches.ts @@ -0,0 +1,5 @@ +import { writerPackage } from "../writer-package.ts"; + +const writer = writerPackage("approaches"); +export const { director, historyCompactor, degenerateTurnCompactor } = writer; +export default writer.workflow; diff --git a/apps/hub/src/workflows/entries/writer-build-plan.ts b/apps/hub/src/workflows/entries/writer-build-plan.ts new file mode 100644 index 00000000..433a23b8 --- /dev/null +++ b/apps/hub/src/workflows/entries/writer-build-plan.ts @@ -0,0 +1,5 @@ +import { writerPackage } from "../writer-package.ts"; + +const writer = writerPackage("build_plan"); +export const { director, historyCompactor, degenerateTurnCompactor } = writer; +export default writer.workflow; diff --git a/apps/hub/src/workflows/entries/writer-constraints.ts b/apps/hub/src/workflows/entries/writer-constraints.ts new file mode 100644 index 00000000..9a8a9bb2 --- /dev/null +++ b/apps/hub/src/workflows/entries/writer-constraints.ts @@ -0,0 +1,5 @@ +import { writerPackage } from "../writer-package.ts"; + +const writer = writerPackage("constraints"); +export const { director, historyCompactor, degenerateTurnCompactor } = writer; +export default writer.workflow; diff --git a/apps/hub/src/workflows/entries/writer-product-requirements.ts b/apps/hub/src/workflows/entries/writer-product-requirements.ts new file mode 100644 index 00000000..8f0e7b34 --- /dev/null +++ b/apps/hub/src/workflows/entries/writer-product-requirements.ts @@ -0,0 +1,5 @@ +import { writerPackage } from "../writer-package.ts"; + +const writer = writerPackage("product_requirements"); +export const { director, historyCompactor, degenerateTurnCompactor } = writer; +export default writer.workflow; diff --git a/apps/hub/src/workflows/entries/writer.ts b/apps/hub/src/workflows/entries/writer.ts deleted file mode 100644 index 9618d110..00000000 --- a/apps/hub/src/workflows/entries/writer.ts +++ /dev/null @@ -1,48 +0,0 @@ -import { createWriterAgent } from "@corbits-builder/writer/agent"; -import { createWriterWorkflow } from "@corbits-builder/writer"; -import { createArtifactTools } from "../../artifacts/tools.ts"; -import { createSkillTools } from "../../skills/tools.ts"; -import { roleInference } from "../role-inference.ts"; -import { compactingDirector } from "../director.ts"; -import { dropDegenerateTurn, trimHistory } from "../compaction.ts"; -import { nativePackageName } from "../package-name.ts"; - -const compacting = compactingDirector(nativePackageName("writer")); - -const { source, contextWindow, parameters } = roleInference("writer"); -const agent = createWriterAgent({ - director: compacting.build({ - contextWindow, - compactAt: 0.6, - fileBeforeDone: true, - }), - id: "document-writer", - capabilities: [], - inference: { ...source, parameters }, - tools: [ - createArtifactTools("writer"), - createSkillTools([ - "constraints", - "approaches", - "product-requirements", - "build-plan", - ]), - ], -}); -const workflow = createWriterWorkflow({ - id: "corbits-builder-writer", - address: "writer", - agent, - inference: { literal: parameters }, - grantRequirements: [ - { resource: "sb-project:*", action: "read", source: "invoker" }, - { resource: "sb-artifact:*", action: "write", source: "invoker" }, - ], -}); -// The directors module of this package: the director the agent names and the compactors it uses. -export const director = compacting.factory; -export const historyCompactor = trimHistory( - contextWindow - parameters.maxTokens, -); -export const degenerateTurnCompactor = dropDegenerateTurn; -export default workflow; diff --git a/apps/hub/src/workflows/install-for-tenant.ts b/apps/hub/src/workflows/install-for-tenant.ts index 21b9702a..e72daeb8 100644 --- a/apps/hub/src/workflows/install-for-tenant.ts +++ b/apps/hub/src/workflows/install-for-tenant.ts @@ -20,6 +20,7 @@ import { import { InstalledWorkflows, INSTALLED_WORKFLOWS_KEY, + roleDeployments, } from "@corbits-builder/schemas/installed-workflows"; import type { RoleBinding } from "@corbits-builder/schemas/models"; import type { WorkflowRole } from "@corbits-builder/schemas/installed-workflows"; @@ -37,6 +38,7 @@ import { import { projectModelChoice } from "../projects/model-choice.ts"; import { ProjectError } from "../projects/errors.ts"; import { buildWorkflowPackages } from "./build-packages.ts"; +import { nativePackageName, WRITER_PACKAGES } from "./package-name.ts"; import { resolveRoleInference, type RoleInference } from "./role-inference.ts"; import type { Actor } from "../lib/actor.ts"; @@ -336,15 +338,15 @@ async function isCurrent( built.map((item) => item.packageName + "@" + item.version), ); if ( - ![...ROLES, "coderAnyPackage" as const].every((key) => { - const entry = installed[key]; - return entry !== undefined && pins.has(entry.pin); - }) + !ROLES.every((role) => + roleDeployments(installed, role).every((entry) => pins.has(entry.pin)), + ) ) return false; const visible = await listVisibleOfferings(db, tenantId); + // A role's packages are installed on the one offering its binding resolved to. for (const role of ROLES) { - const entry = installed[role]; + const [entry] = roleDeployments(installed, role); if (!entry) return false; const row = visible.find( (candidate) => candidate.offering.id === entry.defaultSourceOfferingId, @@ -504,7 +506,12 @@ export async function installForTenant( "@corbits-builder/coordinator-native", "coordinator", ), - writer: deployment("@corbits-builder/writer-native", "writer"), + writers: Object.fromEntries( + Object.entries(WRITER_PACKAGES).map(([kind, name]) => [ + kind, + deployment(nativePackageName(name), "writer"), + ]), + ), reviewer: deployment("@corbits-builder/reviewer-native", "reviewer"), coder: deployment("@corbits-builder/coder-native", "coder"), coderAnyPackage: deployment( diff --git a/apps/hub/src/workflows/package-name.ts b/apps/hub/src/workflows/package-name.ts index 8169fbde..42dbde4f 100644 --- a/apps/hub/src/workflows/package-name.ts +++ b/apps/hub/src/workflows/package-name.ts @@ -1,4 +1,14 @@ +import type { WriterDocumentKind } from "@corbits-builder/schemas/artifact"; + // The npm name each role's workflow package publishes under. A director the package ships is named // under it, since Interchange resolves a director id by its package-name prefix. export const nativePackageName = (role: string) => `@corbits-builder/${role}-native`; + +// Each writer package saves one document kind; entries/ holds one entry per name. +export const WRITER_PACKAGES = { + constraints: "writer-constraints", + approaches: "writer-approaches", + product_requirements: "writer-product-requirements", + build_plan: "writer-build-plan", +} as const satisfies Record; diff --git a/apps/hub/src/workflows/packaged.test.ts b/apps/hub/src/workflows/packaged.test.ts index 24743438..49abd4cb 100644 --- a/apps/hub/src/workflows/packaged.test.ts +++ b/apps/hub/src/workflows/packaged.test.ts @@ -35,7 +35,12 @@ test("native specialist accepts installed tarballs only with an exact package pi test("production entries expose concrete native workflows without callback credentials", async () => { process.env.BUILDER_ROLE_INFERENCE = JSON.stringify(inference.coordinator); const coordinator = (await import("./entries/coordinator.ts")).default; - const writer = (await import("./entries/writer.ts")).default; + const writers = [ + (await import("./entries/writer-constraints.ts")).default, + (await import("./entries/writer-approaches.ts")).default, + (await import("./entries/writer-product-requirements.ts")).default, + (await import("./entries/writer-build-plan.ts")).default, + ]; const reviewer = (await import("./entries/reviewer.ts")).default; const coder = (await import("./entries/coder.ts")).default; const coderAnyPackage = (await import("./entries/coder-any-package.ts")) @@ -43,7 +48,8 @@ test("production entries expose concrete native workflows without callback crede const designer = (await import("./entries/designer.ts")).default; const presenter = (await import("./entries/presenter.ts")).default; expect(coordinator.steps.coordinate).toMatchObject({ triggers: "unbounded" }); - expect(writer.steps.write).toMatchObject({ triggers: 1 }); + for (const writer of writers) + expect(writer.steps.write).toMatchObject({ triggers: 1 }); expect(reviewer.steps.review).toMatchObject({ triggers: 1 }); expect(coder.steps.build).toMatchObject({ triggers: "unbounded" }); expect(coderAnyPackage.steps.build).toMatchObject({ triggers: "unbounded" }); @@ -65,7 +71,7 @@ test("production entries expose concrete native workflows without callback crede expect(installTool(coderAnyPackage)).toEqual({ name: "install_packages" }); for (const workflow of [ coordinator, - writer, + ...writers, reviewer, coder, coderAnyPackage, @@ -76,7 +82,7 @@ test("production entries expose concrete native workflows without callback crede expect( JSON.stringify([ coordinator, - writer, + ...writers, reviewer, coder, coderAnyPackage, @@ -97,7 +103,7 @@ test("built npm tarballs contain importable production workflow metadata and def ); try { const manifest = await buildWorkflowPackages(directory, "1.0.0", inference); - expect(manifest).toHaveLength(7); + expect(manifest).toHaveLength(10); for (const metadata of manifest) { const extracted = join(directory, metadata.assetName); await mkdir(extracted); diff --git a/apps/hub/src/workflows/writer-package.ts b/apps/hub/src/workflows/writer-package.ts new file mode 100644 index 00000000..c575e9c1 --- /dev/null +++ b/apps/hub/src/workflows/writer-package.ts @@ -0,0 +1,51 @@ +import { createWriterAgent } from "@corbits-builder/writer/agent"; +import { createWriterWorkflow } from "@corbits-builder/writer"; +import type { WriterDocumentKind } from "@corbits-builder/schemas/artifact"; +import { createWriterArtifactTools } from "../artifacts/tools.ts"; +import { createSkillTools } from "../skills/tools.ts"; +import { roleInference } from "./role-inference.ts"; +import { compactingDirector } from "./director.ts"; +import { dropDegenerateTurn, trimHistory } from "./compaction.ts"; +import { nativePackageName, WRITER_PACKAGES } from "./package-name.ts"; + +/** The writer's workflow package for one document kind, whose write_artifact saves only that kind. */ +export function writerPackage(kind: WriterDocumentKind) { + const name = WRITER_PACKAGES[kind]; + const compacting = compactingDirector(nativePackageName(name)); + const { source, contextWindow, parameters } = roleInference("writer"); + const agent = createWriterAgent({ + director: compacting.build({ + contextWindow, + compactAt: 0.6, + fileBeforeDone: true, + }), + id: "document-writer", + capabilities: [], + inference: { ...source, parameters }, + tools: [ + createWriterArtifactTools(kind), + createSkillTools([ + "constraints", + "approaches", + "product-requirements", + "build-plan", + ]), + ], + }); + return { + workflow: createWriterWorkflow({ + id: "corbits-builder-" + name, + address: "writer", + agent, + inference: { literal: parameters }, + grantRequirements: [ + { resource: "sb-project:*", action: "read", source: "invoker" }, + { resource: "sb-artifact:*", action: "write", source: "invoker" }, + ], + }), + // The directors module of this package: the director the agent names and the compactors it uses. + director: compacting.factory, + historyCompactor: trimHistory(contextWindow - parameters.maxTokens), + degenerateTurnCompactor: dropDegenerateTurn, + }; +} diff --git a/packages/schemas/src/artifact.ts b/packages/schemas/src/artifact.ts index b0756ee2..71808b55 100644 --- a/packages/schemas/src/artifact.ts +++ b/packages/schemas/src/artifact.ts @@ -10,9 +10,10 @@ const sha256 = /^[a-f0-9]{64}$/; // Documents a model writes as text. A design concept's option is filed from the designer's // workspace by file_design_option, the design by the person's choice among the options, and the // app from the coder's workspace each time check-app passes. -const WriterDocumentKind = type( +export const WriterDocumentKind = type( "'constraints' | 'approaches' | 'product_requirements' | 'build_plan'", ); +export type WriterDocumentKind = typeof WriterDocumentKind.infer; const DocumentKind = WriterDocumentKind.or("'problem_brief'"); const WritableArtifactKind = DocumentKind.or( "'design' | 'design_option' | 'app' | 'approval_package'", @@ -42,36 +43,23 @@ export const BriefWrite = type({ "openQuestions?": OpenQuestions, }); export type BriefWrite = typeof BriefWrite.infer; -export const WriterDocumentWrite = type({ - "+": "reject", - kind: "'constraints'", +const documentFields = { title: text, content, "openQuestions?": OpenQuestions, -}) - .or({ - "+": "reject", - kind: "'product_requirements'", - title: text, - content, - "openQuestions?": OpenQuestions, - }) - .or({ - "+": "reject", - kind: "'build_plan'", - title: text, - content, - "openQuestions?": OpenQuestions, - }) - .or({ +}; +// The fields a writer saves, by the kind it was deployed for; the hub adds the kind. A flat object, +// since some providers read only a tool schema's top-level properties and send nothing for a union. +export const WriterDocumentFields = { + constraints: type({ "+": "reject", ...documentFields }), + approaches: type({ "+": "reject", - kind: "'approaches'", - title: text, - content, - "openQuestions?": OpenQuestions, + ...documentFields, "approaches?": ApproachesMeta, - }); -export type WriterDocumentWrite = typeof WriterDocumentWrite.infer; + }), + product_requirements: type({ "+": "reject", ...documentFields }), + build_plan: type({ "+": "reject", ...documentFields }), +}; export const DocumentWrite = type({ "+": "reject", kind: DocumentKind, diff --git a/packages/schemas/src/installed-workflows.ts b/packages/schemas/src/installed-workflows.ts index 92118d8e..fc2673b5 100644 --- a/packages/schemas/src/installed-workflows.ts +++ b/packages/schemas/src/installed-workflows.ts @@ -5,11 +5,19 @@ import { WorkflowRole } from "./roles.ts"; export { SpawnedRole, WorkflowRole } from "./roles.ts"; export const INSTALLED_WORKFLOWS_KEY = "sb.installedWorkflows"; +// One writer package per document kind, each saving only its kind. +const WriterDeployments = type({ + "+": "reject", + constraints: NativeSpecialistDeployment, + approaches: NativeSpecialistDeployment, + product_requirements: NativeSpecialistDeployment, + build_plan: NativeSpecialistDeployment, +}); export const InstalledWorkflows = type({ "+": "reject", schemaVersion: "1", coordinator: NativeSpecialistDeployment, - writer: NativeSpecialistDeployment, + writers: WriterDeployments, reviewer: NativeSpecialistDeployment, coder: NativeSpecialistDeployment, // The coder whose install_packages does not ask, for a project that turned asking off. @@ -19,6 +27,26 @@ export const InstalledWorkflows = type({ }); export type InstalledWorkflows = typeof InstalledWorkflows.infer; +// Every installed package a run of each role may run on. +const DEPLOYMENTS_BY_ROLE: Record< + WorkflowRole, + (installed: InstalledWorkflows) => NativeSpecialistDeployment[] +> = { + coordinator: (installed) => [installed.coordinator], + writer: (installed) => Object.values(installed.writers), + reviewer: (installed) => [installed.reviewer], + coder: (installed) => [installed.coder, installed.coderAnyPackage], + designer: (installed) => [installed.designer], + presenter: (installed) => [installed.presenter], +}; + +export function roleDeployments( + installed: InstalledWorkflows, + role: WorkflowRole, +) { + return DEPLOYMENTS_BY_ROLE[role](installed); +} + // Returned only to an authenticated native run. Scope comes from the protected // coordinator association/native deployment invoker, never a tool argument. export const WorkflowProjectScope = type({ @@ -29,7 +57,7 @@ export const WorkflowProjectScope = type({ role: WorkflowRole, workflows: { "+": "reject", - writer: NativeSpecialistDeployment, + writers: WriterDeployments, reviewer: NativeSpecialistDeployment, coder: NativeSpecialistDeployment, designer: NativeSpecialistDeployment, diff --git a/test/e2e/oauth-model-login.test.ts b/test/e2e/oauth-model-login.test.ts index 78d45ac6..d53a78cb 100644 --- a/test/e2e/oauth-model-login.test.ts +++ b/test/e2e/oauth-model-login.test.ts @@ -358,8 +358,7 @@ test("ChatGPT and xAI logins are providers a profile binds roles to, and install projectId, ); const visible = await listVisibleOfferings(handle.db, tenantId); - const served = (role: keyof typeof installed) => { - if (role === "schemaVersion") throw new Error("not a role"); + const served = (role: "coder" | "coordinator") => { const row = visible.find( (entry) => entry.offering.id === installed[role].defaultSourceOfferingId, diff --git a/workflows/writer/README.md b/workflows/writer/README.md index 266430fd..f7063a6c 100644 --- a/workflows/writer/README.md +++ b/workflows/writer/README.md @@ -8,11 +8,12 @@ coordinator hands it. This package only declares data. It imports `@intx/agent/authoring` and `@intx/workflow/definition` from Interchange, plus arktype. The -[hub's writer entry][entry] imports it, supplies the model, director and -tools, and builds it into the workflow the [coordinator][coordinator] starts -with `deploy_writer`. The writer's tools are the hub's `write_artifact` and -`read_skill`; it has no read tool, because its task carries every document it -needs in full. +[hub's writer package][entry] imports it, supplies the model, director and +tools, and builds one workflow per document kind; the +[coordinator][coordinator]'s `deploy_writer` starts the one for the kind it +asks for. The writer's tools are the hub's `write_artifact`, which saves only +its own kind's fields, and `read_skill`; it has no read tool, because its task +carries every document it needs in full. ## Layout @@ -34,6 +35,6 @@ What a document should contain lives in its skill rather than the prompt; prompt conventions are in [AGENTS.md][agents]. `bun run typecheck` and `bun run test` cover it, and `bun run dev` runs it in a real project. -[entry]: https://github.com/corbitsdev/corbits-builder/blob/main/apps/hub/src/workflows/entries/writer.ts +[entry]: https://github.com/corbitsdev/corbits-builder/blob/main/apps/hub/src/workflows/writer-package.ts [coordinator]: https://github.com/corbitsdev/corbits-builder/tree/main/workflows/coordinator [agents]: https://github.com/corbitsdev/corbits-builder/blob/main/AGENTS.md From f41901a1ae2aeb28410a84a0f342403ec3a88f36 Mon Sep 17 00:00:00 2001 From: Sawyer Date: Wed, 4 Nov 2026 00:31:03 -0800 Subject: [PATCH 2/5] fix(hub): keep old bundles and echoed kinds working with split writers Writers accept a kind matching their own, which models echo from the task. The scope keeps writer for non-coordinator bundles frozen before the split. deploy_writer offers only writer kinds, and each role's packages must share one offering to count as current. --- apps/hub/src/artifacts/tools.test.ts | 71 ++++++++++---------- apps/hub/src/artifacts/tools.ts | 19 ++++-- apps/hub/src/auth/agent-token-bridge.ts | 6 ++ apps/hub/src/projects/coordinator-upgrade.ts | 2 +- apps/hub/src/projects/deploy-tools.test.ts | 6 +- apps/hub/src/projects/deploy-tools.ts | 28 ++------ apps/hub/src/workflows/install-for-tenant.ts | 11 ++- packages/schemas/src/artifact.ts | 24 +++++-- packages/schemas/src/installed-workflows.ts | 2 + 9 files changed, 94 insertions(+), 75 deletions(-) diff --git a/apps/hub/src/artifacts/tools.test.ts b/apps/hub/src/artifacts/tools.test.ts index dcbf9016..409ae00d 100644 --- a/apps/hub/src/artifacts/tools.test.ts +++ b/apps/hub/src/artifacts/tools.test.ts @@ -45,10 +45,10 @@ const sampleApproaches = { // The fields each writer's save takes. const WRITER_FIELDS: Record = { - constraints: ["content", "openQuestions", "title"], - approaches: ["approaches", "content", "openQuestions", "title"], - product_requirements: ["content", "openQuestions", "title"], - build_plan: ["content", "openQuestions", "title"], + constraints: ["content", "kind", "openQuestions", "title"], + approaches: ["approaches", "content", "kind", "openQuestions", "title"], + product_requirements: ["content", "kind", "openQuestions", "title"], + build_plan: ["content", "kind", "openQuestions", "title"], }; const WRITER_KINDS = Object.keys(WRITER_FIELDS).map((kind) => WriterDocumentKind.assert(kind), @@ -199,38 +199,39 @@ test("each writer's write_artifact lists only its own fields, at the top level", } }); -test("a writer saves without a kind and files under the kind it was deployed for", async () => { - const constraints = harness( - "writer", - createWriterArtifactTools("constraints"), - ); - const named = await constraints.write({ - kind: "build_plan", - title: "Build Plan", - content: "# Build Plan", - }); - expect(named.isError).toBe(true); - expect(named.content).toStartWith("invalid_arguments:"); - const withApproaches = await constraints.write({ - title: "Constraints", - content: "# Constraints", - approaches: sampleApproaches, - }); - expect(withApproaches.isError).toBe(true); - expect(withApproaches.content).toContain("approaches"); - expect( - (await constraints.write({ title: "Constraints", content: "" })).isError, - ).toBe(true); - expect(constraints.written).toHaveLength(0); +test("a writer files under the kind it was deployed for, with or without naming it", async () => { + for (const kind of WRITER_KINDS) { + const writer = harness("writer", createWriterArtifactTools(kind)); + const otherKind = await writer.write({ + kind: WRITER_KINDS.find((other) => other !== kind), + title: "Doc", + content: "# Doc", + }); + expect(otherKind.content).toStartWith("invalid_arguments:"); + const empty = await writer.write({ title: "Doc", content: "" }); + expect(empty.content).toStartWith("invalid_arguments:"); + expect(writer.written).toHaveLength(0); - const saved = await constraints.write({ - title: "Constraints", - content: "# Constraints", - }); - expect(saved.isError).toBeUndefined(); - expect(constraints.written).toEqual([ - { kind: "constraints", title: "Constraints", content: "# Constraints" }, - ]); + await writer.write({ title: "Unnamed", content: "# Doc" }); + await writer.write({ kind, title: "Named", content: "# Doc" }); + expect(writer.written).toEqual([ + { kind, title: "Unnamed", content: "# Doc" }, + { kind, title: "Named", content: "# Doc" }, + ]); + } +}); + +test("only the approaches writer saves approaches", async () => { + for (const kind of WRITER_KINDS.filter((kind) => kind !== "approaches")) { + const writer = harness("writer", createWriterArtifactTools(kind)); + const result = await writer.write({ + title: "Doc", + content: "# Doc", + approaches: sampleApproaches, + }); + expect(result.content).toStartWith("invalid_arguments:"); + expect(result.content).toContain("approaches"); + } const approaches = harness("writer", createWriterArtifactTools("approaches")); const valid = await approaches.write({ diff --git a/apps/hub/src/artifacts/tools.ts b/apps/hub/src/artifacts/tools.ts index 4b57d388..584b23e4 100644 --- a/apps/hub/src/artifacts/tools.ts +++ b/apps/hub/src/artifacts/tools.ts @@ -59,23 +59,28 @@ const briefWriteDefinition = { // Each writer's document by name, and what its write_artifact says about the fields only it has. const WRITER_DOCUMENTS: Record< WriterDocumentKind, - { name: string; fieldNotes: string } + { name: string; fieldNotes: string[] } > = { - constraints: { name: "constraints", fieldNotes: "" }, + constraints: { name: "constraints", fieldNotes: [] }, approaches: { name: "approaches document", - fieldNotes: - "\napproaches carries the approaches the document compares, for the person to see side by side: each with a short id, its name, a one-line summary, bestFor, its pros, cons and risks as short phrases, and its effort; recommended names the id of the one you recommend.", + fieldNotes: [ + "approaches carries the approaches the document compares, for the person to see side by side: each with a short id, its name, a one-line summary, bestFor, its pros, cons and risks as short phrases, and its effort; recommended names the id of the one you recommend.", + ], }, - product_requirements: { name: "product requirements", fieldNotes: "" }, - build_plan: { name: "build plan", fieldNotes: "" }, + product_requirements: { name: "product requirements", fieldNotes: [] }, + build_plan: { name: "build plan", fieldNotes: [] }, }; function writerWriteDefinition(kind: WriterDocumentKind) { const { name, fieldNotes } = WRITER_DOCUMENTS[kind]; return { name: "write_artifact", - description: `Write the next version of the ${name}. The hub assigns its id, version and sha256 and returns them. Writing a version is not approval.\nopenQuestions lists each question in the document's Open questions section that the person must answer: a short id, the question as they would read it, whyItBlocks, and suggestedAnswers, your recommendation first. A version written without openQuestions leaves none open.${fieldNotes}`, + description: [ + `Write the next version of the ${name}. The hub assigns its id, version and sha256 and returns them. Writing a version is not approval.`, + "openQuestions lists each question in the document's Open questions section that the person must answer: a short id, the question as they would read it, whyItBlocks, and suggestedAnswers, your recommendation first. A version written without openQuestions leaves none open.", + ...fieldNotes, + ].join("\n"), inputSchema: { ...WriterDocumentFields[kind].toJsonSchema() }, }; } diff --git a/apps/hub/src/auth/agent-token-bridge.ts b/apps/hub/src/auth/agent-token-bridge.ts index e6af601b..de11a5db 100644 --- a/apps/hub/src/auth/agent-token-bridge.ts +++ b/apps/hub/src/auth/agent-token-bridge.ts @@ -294,6 +294,12 @@ export function createAgentTokenBridge( role, workflows: { writers: installed.writers, + // A bundle frozen before writers were split requires writer to read its scope. Only a + // coordinator deploys from it, and an old one would start the constraints writer for any + // kind, so a coordinator goes without until it is upgraded. + ...(role === "coordinator" + ? {} + : { writer: installed.writers.constraints }), reviewer: installed.reviewer, coder: installsAnyPackage(preferences.value) ? installed.coderAnyPackage diff --git a/apps/hub/src/projects/coordinator-upgrade.ts b/apps/hub/src/projects/coordinator-upgrade.ts index 0c536f96..7274dd14 100644 --- a/apps/hub/src/projects/coordinator-upgrade.ts +++ b/apps/hub/src/projects/coordinator-upgrade.ts @@ -52,7 +52,7 @@ const pinOf = (bundle: unknown) => { }; // The pins a role may run on: a coder is either coder package, by the project's package choice, and a -// writer the package for the kind it was deployed for. +// writer any writer package, one per document kind. const pinsFor = (installed: InstalledWorkflows, role: WorkflowRole) => roleDeployments(installed, role).map((deployment) => deployment.pin); diff --git a/apps/hub/src/projects/deploy-tools.test.ts b/apps/hub/src/projects/deploy-tools.test.ts index a92540e5..3557e1b9 100644 --- a/apps/hub/src/projects/deploy-tools.test.ts +++ b/apps/hub/src/projects/deploy-tools.test.ts @@ -335,11 +335,11 @@ test("one ask approves several documents, and a writer starts once every earlier ]); }); -test("deploy_writer has no writer for the reviewer's engineering review", async () => { +test("deploy_writer offers only the kinds a writer saves", async () => { const { run, deployed } = writerTools({ agent_decides_approvals: "yes" }); expect( - (await run("deploy_writer", writer("engineering_review"))).content, - ).toStartWith("no_writer_for_kind:"); + (await run("deploy_writer", writer("engineering_review"))).isError, + ).toBe(true); expect(deployed).toEqual([]); }); diff --git a/apps/hub/src/projects/deploy-tools.ts b/apps/hub/src/projects/deploy-tools.ts index fa59f030..051df432 100644 --- a/apps/hub/src/projects/deploy-tools.ts +++ b/apps/hub/src/projects/deploy-tools.ts @@ -18,6 +18,7 @@ import { NativeSpecialistDeployment, } from "@corbits-builder/schemas/native-specialist"; import { SourceRef } from "@corbits-builder/schemas"; +import type { WorkflowProjectScope } from "@corbits-builder/schemas/installed-workflows"; import { ArtifactContent, DesignBundle, @@ -70,7 +71,7 @@ const Instruction = NativeSpecialistAssignment.get("instruction").and("string >= 40"); const DeployWriter = type({ "+": "reject", - kind: NativeDocumentKind, + kind: WriterDocumentKind, instruction: Instruction, }); const APPROVABLE = [ @@ -236,13 +237,7 @@ export function createDeployTools(options?: { tenantId: string; projectId: string; domain: string; - workflows: { - writers: Record; - reviewer: NativeSpecialistDeployment; - coder: NativeSpecialistDeployment; - designer: NativeSpecialistDeployment; - presenter?: NativeSpecialistDeployment; - }; + workflows: WorkflowProjectScope["workflows"]; maxResultBytes: number; }): AnnotatedToolFactory { if (options) { @@ -539,21 +534,12 @@ export function createDeployTools(options?: { result = { approved: documents }; } else if (call.name === "deploy_writer") { const input = validArguments(DeployWriter, call.arguments); - const writerKind = WriterDocumentKind(input.kind); - if (writerKind instanceof type.errors) - throw new Refusal( - "no_writer_for_kind: review_artifact has the reviewer write the " + - input.kind, - ); const { byKind, approved, preferences } = await listed(); const briefRef = required(byKind, "problem_brief"); - const position = DOCUMENT_ORDER.indexOf( - input.kind as (typeof DOCUMENT_ORDER)[number], + const earlier = DOCUMENT_ORDER.slice( + 0, + DOCUMENT_ORDER.indexOf(input.kind), ); - const earlier = - position < 0 - ? DOCUMENT_ORDER.slice(0, 1) - : DOCUMENT_ORDER.slice(0, position); // An imported project may lack an earlier document that its later ones were written without. const writtenWithout = (index: number) => DOCUMENT_ORDER.slice(index + 1).some((kind) => byKind.has(kind)); @@ -587,7 +573,7 @@ export function createDeployTools(options?: { .map(([, ref]) => ref); result = await deploy( "writer", - options.workflows.writers[writerKind], + options.workflows.writers[input.kind], preferences, "writer-" + input.kind, await assignmentToWriterMessage( diff --git a/apps/hub/src/workflows/install-for-tenant.ts b/apps/hub/src/workflows/install-for-tenant.ts index e72daeb8..c3ac8de5 100644 --- a/apps/hub/src/workflows/install-for-tenant.ts +++ b/apps/hub/src/workflows/install-for-tenant.ts @@ -346,10 +346,15 @@ async function isCurrent( const visible = await listVisibleOfferings(db, tenantId); // A role's packages are installed on the one offering its binding resolved to. for (const role of ROLES) { - const [entry] = roleDeployments(installed, role); - if (!entry) return false; + const offeringIds = new Set( + roleDeployments(installed, role).map( + (entry) => entry.defaultSourceOfferingId, + ), + ); + if (offeringIds.size !== 1) return false; + const [offeringId] = offeringIds; const row = visible.find( - (candidate) => candidate.offering.id === entry.defaultSourceOfferingId, + (candidate) => candidate.offering.id === offeringId, ); if (!row || !(await servesSource(db, directory, row, bindings[role]))) return false; diff --git a/packages/schemas/src/artifact.ts b/packages/schemas/src/artifact.ts index 71808b55..86af425f 100644 --- a/packages/schemas/src/artifact.ts +++ b/packages/schemas/src/artifact.ts @@ -48,17 +48,31 @@ const documentFields = { content, "openQuestions?": OpenQuestions, }; -// The fields a writer saves, by the kind it was deployed for; the hub adds the kind. A flat object, -// since some providers read only a tool schema's top-level properties and send nothing for a union. +// The fields a writer saves, by the kind it was deployed for; the hub files it under that kind, and a +// kind the writer repeats from its task must match. A flat object, since some providers read only a +// tool schema's top-level properties and send nothing for a union. export const WriterDocumentFields = { - constraints: type({ "+": "reject", ...documentFields }), + constraints: type({ + "+": "reject", + "kind?": "'constraints'", + ...documentFields, + }), approaches: type({ "+": "reject", + "kind?": "'approaches'", ...documentFields, "approaches?": ApproachesMeta, }), - product_requirements: type({ "+": "reject", ...documentFields }), - build_plan: type({ "+": "reject", ...documentFields }), + product_requirements: type({ + "+": "reject", + "kind?": "'product_requirements'", + ...documentFields, + }), + build_plan: type({ + "+": "reject", + "kind?": "'build_plan'", + ...documentFields, + }), }; export const DocumentWrite = type({ "+": "reject", diff --git a/packages/schemas/src/installed-workflows.ts b/packages/schemas/src/installed-workflows.ts index fc2673b5..fcd84e66 100644 --- a/packages/schemas/src/installed-workflows.ts +++ b/packages/schemas/src/installed-workflows.ts @@ -58,6 +58,8 @@ export const WorkflowProjectScope = type({ workflows: { "+": "reject", writers: WriterDeployments, + // Read only by bundles frozen before writers were split; nothing deploys from it now. + "writer?": NativeSpecialistDeployment, reviewer: NativeSpecialistDeployment, coder: NativeSpecialistDeployment, designer: NativeSpecialistDeployment, From a3297191589b498c7e2c32c95f494b68f66f6687 Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Wed, 4 Nov 2026 11:32:42 -0800 Subject: [PATCH 3/5] refactor(hub): key installed workflows by package Each writer package has its own key in the install record and project scope, so one key returns its deployment; a single package table maps each package to its role for build, install and upgrade. An unreadable install record now reinstalls the tenant instead of refusing every scope read, and the up-to-date check matches each package against its own offering. A contract test keeps the scope readable by specialists frozen on the last release. --- apps/hub/src/app.ts | 11 ++- .../src/artifacts/approval-package.test.ts | 10 ++- apps/hub/src/artifacts/design-draft.test.ts | 20 +++--- apps/hub/src/artifacts/tools.test.ts | 10 ++- apps/hub/src/auth/agent-token-bridge.ts | 39 +++++----- .../src/projects/coordinator-upgrade.test.ts | 32 ++++----- .../hub/src/projects/deploy-presenter.test.ts | 20 +++--- apps/hub/src/projects/deploy-tools.test.ts | 12 ++-- apps/hub/src/projects/deploy-tools.ts | 13 ++-- apps/hub/src/projects/errors.test.ts | 10 ++- apps/hub/src/workflows/build-packages.ts | 26 +++---- apps/hub/src/workflows/director.test.ts | 10 ++- apps/hub/src/workflows/install-for-tenant.ts | 53 +++++--------- apps/hub/src/workflows/package-name.ts | 10 --- apps/hub/src/workflows/project-scope.test.ts | 51 +++++++++++++ apps/hub/src/workflows/project-scope.ts | 27 +++++++ apps/hub/src/workflows/writer-package.ts | 3 +- packages/schemas/src/installed-workflows.ts | 72 ++++++++++++------- 18 files changed, 243 insertions(+), 186 deletions(-) create mode 100644 apps/hub/src/workflows/project-scope.test.ts diff --git a/apps/hub/src/app.ts b/apps/hub/src/app.ts index 7645707b..99e1ba86 100644 --- a/apps/hub/src/app.ts +++ b/apps/hub/src/app.ts @@ -419,7 +419,16 @@ export async function createHubRuntime( app.use(liveAppProxy.refuseAppOrigins); app.use(createHubContextMiddleware({ getSession })); app.use(gitTokens.middleware); - const coordinatorTokens = createAgentTokenBridge(handle.db, runTokenStore); + const coordinatorTokens = createAgentTokenBridge( + handle.db, + runTokenStore, + (actor, projectId) => + installForTenant( + { db: handle.db, assets: assetService }, + actor, + projectId, + ), + ); app.use("/api/tenants/:tenantId/*", coordinatorTokens.middleware); app.use("/api/workflow-project-scope", coordinatorTokens.middleware); app.use( diff --git a/apps/hub/src/artifacts/approval-package.test.ts b/apps/hub/src/artifacts/approval-package.test.ts index ac729769..9b9c58f6 100644 --- a/apps/hub/src/artifacts/approval-package.test.ts +++ b/apps/hub/src/artifacts/approval-package.test.ts @@ -345,12 +345,10 @@ describe("approval package validation and secret scanning", () => { domain: "example.invalid", role: "coordinator", workflows: { - writers: { - constraints: deployment, - approaches: deployment, - product_requirements: deployment, - build_plan: deployment, - }, + "writer-constraints": deployment, + "writer-approaches": deployment, + "writer-product-requirements": deployment, + "writer-build-plan": deployment, reviewer: deployment, coder: deployment, designer: deployment, diff --git a/apps/hub/src/artifacts/design-draft.test.ts b/apps/hub/src/artifacts/design-draft.test.ts index 3cd4469f..e31cde00 100644 --- a/apps/hub/src/artifacts/design-draft.test.ts +++ b/apps/hub/src/artifacts/design-draft.test.ts @@ -175,12 +175,10 @@ test("file_design_option packs per-screen files and drops the untouched starter domain: "example.invalid", role: "designer", workflows: { - writers: { - constraints: deployment, - approaches: deployment, - product_requirements: deployment, - build_plan: deployment, - }, + "writer-constraints": deployment, + "writer-approaches": deployment, + "writer-product-requirements": deployment, + "writer-build-plan": deployment, reviewer: deployment, coder: deployment, designer: deployment, @@ -461,12 +459,10 @@ test("designer agent writes per-screen files, is held by checkBeforeDone until c domain: "example.invalid", role: "designer", workflows: { - writers: { - constraints: deployment, - approaches: deployment, - product_requirements: deployment, - build_plan: deployment, - }, + "writer-constraints": deployment, + "writer-approaches": deployment, + "writer-product-requirements": deployment, + "writer-build-plan": deployment, reviewer: deployment, coder: deployment, designer: deployment, diff --git a/apps/hub/src/artifacts/tools.test.ts b/apps/hub/src/artifacts/tools.test.ts index 409ae00d..6b3ea8c9 100644 --- a/apps/hub/src/artifacts/tools.test.ts +++ b/apps/hub/src/artifacts/tools.test.ts @@ -74,12 +74,10 @@ function harness( domain: "example.invalid", role, workflows: { - writers: { - constraints: deployment, - approaches: deployment, - product_requirements: deployment, - build_plan: deployment, - }, + "writer-constraints": deployment, + "writer-approaches": deployment, + "writer-product-requirements": deployment, + "writer-build-plan": deployment, reviewer: deployment, coder: deployment, designer: deployment, diff --git a/apps/hub/src/auth/agent-token-bridge.ts b/apps/hub/src/auth/agent-token-bridge.ts index de11a5db..b30aa7a6 100644 --- a/apps/hub/src/auth/agent-token-bridge.ts +++ b/apps/hub/src/auth/agent-token-bridge.ts @@ -1,3 +1,4 @@ +import { type } from "arktype"; import { InstalledWorkflows, INSTALLED_WORKFLOWS_KEY, @@ -24,10 +25,12 @@ import { import { and, eq } from "drizzle-orm"; import type { MiddlewareHandler } from "hono"; import { getLogger } from "@intx/log"; +import type { Actor } from "../lib/actor.ts"; import { createCoordinatorRegistry } from "./coordinator-registry.ts"; import { projectResource } from "../projects/errors.ts"; import { spawnedWorkflowRole } from "../workflows/spawn-role.ts"; import { buildPackageLayout } from "../builds/package-layout.ts"; +import { scopeWorkflows } from "../workflows/project-scope.ts"; import { installsAnyPackage, opensNetwork } from "../config/preferences.ts"; import { loadSettings } from "../settings/policy.ts"; @@ -167,6 +170,8 @@ async function nativeRunIdentity( export function createAgentTokenBridge( db: ReturnType["db"], runTokens?: RunTokenLookup, + // Reinstalls a tenant whose stored packages predate this hub, so a run is never refused for it. + install?: (actor: Actor, projectId: string) => Promise, ) { const registry = createCoordinatorRegistry(db); const authenticated = new WeakSet(); @@ -279,9 +284,17 @@ export function createAgentTokenBridge( }); if (!row) return denied(); const config = row.config as Record | null; - const installed = InstalledWorkflows.assert( - config?.[INSTALLED_WORKFLOWS_KEY], - ); + const stored = InstalledWorkflows(config?.[INSTALLED_WORKFLOWS_KEY]); + const installed = + stored instanceof type.errors + ? install && + binding.run.principalId && + (await install( + { tenantId: row.id, principalId: binding.run.principalId }, + binding.projectId, + )) + : stored; + if (!installed) return denied(); const { preferences } = await loadSettings( db, identity.tenantId, @@ -292,21 +305,11 @@ export function createAgentTokenBridge( domain: row.domain, projectId: binding.projectId, role, - workflows: { - writers: installed.writers, - // A bundle frozen before writers were split requires writer to read its scope. Only a - // coordinator deploys from it, and an old one would start the constraints writer for any - // kind, so a coordinator goes without until it is upgraded. - ...(role === "coordinator" - ? {} - : { writer: installed.writers.constraints }), - reviewer: installed.reviewer, - coder: installsAnyPackage(preferences.value) - ? installed.coderAnyPackage - : installed.coder, - designer: installed.designer, - ...(installed.presenter ? { presenter: installed.presenter } : {}), - }, + workflows: scopeWorkflows( + installed, + role, + installsAnyPackage(preferences.value), + ), openNetwork: opensNetwork(preferences.value), build: await buildLayout(), }); diff --git a/apps/hub/src/projects/coordinator-upgrade.test.ts b/apps/hub/src/projects/coordinator-upgrade.test.ts index 49fc02de..30f4197a 100644 --- a/apps/hub/src/projects/coordinator-upgrade.test.ts +++ b/apps/hub/src/projects/coordinator-upgrade.test.ts @@ -14,25 +14,25 @@ const deployment = (pin: string) => ({ defaultSourceOfferingId: "offering", }); const installed: InstalledWorkflows = { - schemaVersion: 1, + schemaVersion: 2, coordinator: deployment("@corbits-builder/coordinator-native@1.0.0-sha-new"), - writers: { - constraints: deployment( - "@corbits-builder/writer-constraints-native@1.0.0-sha-new", - ), - approaches: deployment( - "@corbits-builder/writer-approaches-native@1.0.0-sha-new", - ), - product_requirements: deployment( - "@corbits-builder/writer-product-requirements-native@1.0.0-sha-new", - ), - build_plan: deployment( - "@corbits-builder/writer-build-plan-native@1.0.0-sha-new", - ), - }, + "writer-constraints": deployment( + "@corbits-builder/writer-constraints-native@1.0.0-sha-new", + ), + "writer-approaches": deployment( + "@corbits-builder/writer-approaches-native@1.0.0-sha-new", + ), + "writer-product-requirements": deployment( + "@corbits-builder/writer-product-requirements-native@1.0.0-sha-new", + ), + "writer-build-plan": deployment( + "@corbits-builder/writer-build-plan-native@1.0.0-sha-new", + ), reviewer: deployment("@corbits-builder/reviewer-native@1.0.0-sha-new"), coder: deployment("@corbits-builder/coder-native@1.0.0-sha-new"), - coderAnyPackage: deployment("@corbits-builder/coder-native@1.0.0-sha-any"), + "coder-any-package": deployment( + "@corbits-builder/coder-native@1.0.0-sha-any", + ), designer: deployment("@corbits-builder/designer-native@1.0.0-sha-new"), presenter: deployment("@corbits-builder/presenter-native@1.0.0-sha-new"), }; diff --git a/apps/hub/src/projects/deploy-presenter.test.ts b/apps/hub/src/projects/deploy-presenter.test.ts index e74153c4..a553f9a9 100644 --- a/apps/hub/src/projects/deploy-presenter.test.ts +++ b/apps/hub/src/projects/deploy-presenter.test.ts @@ -200,12 +200,10 @@ test("deploy_presenter requires approved documents and pins source refs and file projectId: "project", domain: "example.invalid", workflows: { - writers: { - constraints: deployment, - approaches: deployment, - product_requirements: deployment, - build_plan: deployment, - }, + "writer-constraints": deployment, + "writer-approaches": deployment, + "writer-product-requirements": deployment, + "writer-build-plan": deployment, reviewer: deployment, coder: deployment, designer: deployment, @@ -286,12 +284,10 @@ test("file_package refuses until check_deck passes on the current deck, then fil domain: "example.invalid", role: "presenter", workflows: { - writers: { - constraints: deployment, - approaches: deployment, - product_requirements: deployment, - build_plan: deployment, - }, + "writer-constraints": deployment, + "writer-approaches": deployment, + "writer-product-requirements": deployment, + "writer-build-plan": deployment, reviewer: deployment, coder: deployment, designer: deployment, diff --git a/apps/hub/src/projects/deploy-tools.test.ts b/apps/hub/src/projects/deploy-tools.test.ts index 3557e1b9..6e0ec6b4 100644 --- a/apps/hub/src/projects/deploy-tools.test.ts +++ b/apps/hub/src/projects/deploy-tools.test.ts @@ -18,12 +18,12 @@ const writerDeployment = (name: string) => ({ pin: `@corbits-builder/${name}-native@1.0.0`, }); const workflows = { - writers: { - constraints: writerDeployment("writer-constraints"), - approaches: writerDeployment("writer-approaches"), - product_requirements: writerDeployment("writer-product-requirements"), - build_plan: writerDeployment("writer-build-plan"), - }, + "writer-constraints": writerDeployment("writer-constraints"), + "writer-approaches": writerDeployment("writer-approaches"), + "writer-product-requirements": writerDeployment( + "writer-product-requirements", + ), + "writer-build-plan": writerDeployment("writer-build-plan"), reviewer: deployment, coder: deployment, designer: deployment, diff --git a/apps/hub/src/projects/deploy-tools.ts b/apps/hub/src/projects/deploy-tools.ts index 051df432..d7f152ec 100644 --- a/apps/hub/src/projects/deploy-tools.ts +++ b/apps/hub/src/projects/deploy-tools.ts @@ -18,7 +18,10 @@ import { NativeSpecialistDeployment, } from "@corbits-builder/schemas/native-specialist"; import { SourceRef } from "@corbits-builder/schemas"; -import type { WorkflowProjectScope } from "@corbits-builder/schemas/installed-workflows"; +import { + WRITER_PACKAGES, + type WorkflowProjectScope, +} from "@corbits-builder/schemas/installed-workflows"; import { ArtifactContent, DesignBundle, @@ -241,11 +244,7 @@ export function createDeployTools(options?: { maxResultBytes: number; }): AnnotatedToolFactory { if (options) { - const { writers, ...specialists } = options.workflows; - for (const deployment of [ - ...Object.values(writers), - ...Object.values(specialists), - ]) { + for (const deployment of Object.values(options.workflows)) { if (!deployment) continue; const checked = NativeSpecialistDeployment.assert(deployment); if (!checked.sourceOfferingIds.includes(checked.defaultSourceOfferingId)) @@ -573,7 +572,7 @@ export function createDeployTools(options?: { .map(([, ref]) => ref); result = await deploy( "writer", - options.workflows.writers[input.kind], + options.workflows[WRITER_PACKAGES[input.kind]], preferences, "writer-" + input.kind, await assignmentToWriterMessage( diff --git a/apps/hub/src/projects/errors.test.ts b/apps/hub/src/projects/errors.test.ts index 75d87325..4c388bdc 100644 --- a/apps/hub/src/projects/errors.test.ts +++ b/apps/hub/src/projects/errors.test.ts @@ -63,12 +63,10 @@ test("a rejected write_artifact call reaches the agent with {path, message} issu domain: "example.invalid", role: "writer", workflows: { - writers: { - constraints: deployment, - approaches: deployment, - product_requirements: deployment, - build_plan: deployment, - }, + "writer-constraints": deployment, + "writer-approaches": deployment, + "writer-product-requirements": deployment, + "writer-build-plan": deployment, reviewer: deployment, coder: deployment, designer: deployment, diff --git a/apps/hub/src/workflows/build-packages.ts b/apps/hub/src/workflows/build-packages.ts index 2a9fa933..adcd1d4a 100644 --- a/apps/hub/src/workflows/build-packages.ts +++ b/apps/hub/src/workflows/build-packages.ts @@ -2,9 +2,12 @@ import { createHash } from "node:crypto"; import { mkdir, mkdtemp, rm, writeFile } from "node:fs/promises"; import { join, resolve } from "node:path"; import { tmpdir } from "node:os"; -import { nativePackageName, WRITER_PACKAGES } from "./package-name.ts"; +import { nativePackageName } from "./package-name.ts"; import type { RoleInference } from "./role-inference.ts"; -import type { WorkflowRole } from "@corbits-builder/schemas/installed-workflows"; +import { + WORKFLOW_PACKAGES, + type WorkflowRole, +} from "@corbits-builder/schemas/installed-workflows"; // The version follows the bundle bytes, so a rebuilt workflow publishes beside the old one instead of // colliding with a tarball the registry already holds. @@ -13,19 +16,10 @@ function contentVersion(base: string, bundle: string): string { } // Each package is an entry under entries/ built on what its role's model binding resolved to. -const WORKFLOW_PACKAGES: readonly { name: string; role: WorkflowRole }[] = [ - ...( - ["coordinator", "reviewer", "coder", "designer", "presenter"] as const - ).map((role) => ({ - name: role, - role, - })), - ...Object.values(WRITER_PACKAGES).map((name) => ({ - name, - role: "writer" as const, - })), - { name: "coder-any-package", role: "coder" }, -]; +const PACKAGES = Object.entries(WORKFLOW_PACKAGES).map(([name, role]) => ({ + name, + role, +})); export async function buildWorkflowPackages( destination: string, @@ -36,7 +30,7 @@ export async function buildWorkflowPackages( const output = resolve(destination); await mkdir(output, { recursive: true }); const manifest = []; - for (const { name, role } of WORKFLOW_PACKAGES) { + for (const { name, role } of PACKAGES) { const result = await Bun.build({ define: { "process.env.BUILDER_ROLE_INFERENCE": JSON.stringify( diff --git a/apps/hub/src/workflows/director.test.ts b/apps/hub/src/workflows/director.test.ts index 22b72e9b..00a74f15 100644 --- a/apps/hub/src/workflows/director.test.ts +++ b/apps/hub/src/workflows/director.test.ts @@ -443,12 +443,10 @@ test("check_design validates DESIGN.md and .dark in styles.css and auto-files th domain: "example.invalid", role: "designer", workflows: { - writers: { - constraints: deployment, - approaches: deployment, - product_requirements: deployment, - build_plan: deployment, - }, + "writer-constraints": deployment, + "writer-approaches": deployment, + "writer-product-requirements": deployment, + "writer-build-plan": deployment, reviewer: deployment, coder: deployment, designer: deployment, diff --git a/apps/hub/src/workflows/install-for-tenant.ts b/apps/hub/src/workflows/install-for-tenant.ts index c3ac8de5..446a18ca 100644 --- a/apps/hub/src/workflows/install-for-tenant.ts +++ b/apps/hub/src/workflows/install-for-tenant.ts @@ -20,7 +20,8 @@ import { import { InstalledWorkflows, INSTALLED_WORKFLOWS_KEY, - roleDeployments, + WORKFLOW_PACKAGES, + type WorkflowPackage, } from "@corbits-builder/schemas/installed-workflows"; import type { RoleBinding } from "@corbits-builder/schemas/models"; import type { WorkflowRole } from "@corbits-builder/schemas/installed-workflows"; @@ -38,7 +39,7 @@ import { import { projectModelChoice } from "../projects/model-choice.ts"; import { ProjectError } from "../projects/errors.ts"; import { buildWorkflowPackages } from "./build-packages.ts"; -import { nativePackageName, WRITER_PACKAGES } from "./package-name.ts"; +import { nativePackageName } from "./package-name.ts"; import { resolveRoleInference, type RoleInference } from "./role-inference.ts"; import type { Actor } from "../lib/actor.ts"; @@ -325,7 +326,7 @@ const ROLES = [ "presenter", ] as const; -/** A stored descriptor is current only while it pins this build's packages and, for each role, an offering serving its source. */ +/** A stored descriptor is current only while it pins this build's packages and each package's offering still serves its role's source. */ async function isCurrent( db: DB["db"], directory: ModelDirectory, @@ -337,24 +338,16 @@ async function isCurrent( const pins = new Set( built.map((item) => item.packageName + "@" + item.version), ); - if ( - !ROLES.every((role) => - roleDeployments(installed, role).every((entry) => pins.has(entry.pin)), - ) - ) - return false; + const packages = Object.entries(WORKFLOW_PACKAGES) as [ + WorkflowPackage, + WorkflowRole, + ][]; + if (!packages.every(([name]) => pins.has(installed[name].pin))) return false; const visible = await listVisibleOfferings(db, tenantId); - // A role's packages are installed on the one offering its binding resolved to. - for (const role of ROLES) { - const offeringIds = new Set( - roleDeployments(installed, role).map( - (entry) => entry.defaultSourceOfferingId, - ), - ); - if (offeringIds.size !== 1) return false; - const [offeringId] = offeringIds; + for (const [name, role] of packages) { const row = visible.find( - (candidate) => candidate.offering.id === offeringId, + (candidate) => + candidate.offering.id === installed[name].defaultSourceOfferingId, ); if (!row || !(await servesSource(db, directory, row, bindings[role]))) return false; @@ -506,25 +499,13 @@ export async function installForTenant( }; }; const descriptor = InstalledWorkflows.assert({ - schemaVersion: 1, - coordinator: deployment( - "@corbits-builder/coordinator-native", - "coordinator", - ), - writers: Object.fromEntries( - Object.entries(WRITER_PACKAGES).map(([kind, name]) => [ - kind, - deployment(nativePackageName(name), "writer"), + schemaVersion: 2, + ...Object.fromEntries( + Object.entries(WORKFLOW_PACKAGES).map(([name, role]) => [ + name, + deployment(nativePackageName(name), role), ]), ), - reviewer: deployment("@corbits-builder/reviewer-native", "reviewer"), - coder: deployment("@corbits-builder/coder-native", "coder"), - coderAnyPackage: deployment( - "@corbits-builder/coder-any-package-native", - "coder", - ), - designer: deployment("@corbits-builder/designer-native", "designer"), - presenter: deployment("@corbits-builder/presenter-native", "presenter"), }); await ports.db.transaction(async (tx) => { const [current] = await tx diff --git a/apps/hub/src/workflows/package-name.ts b/apps/hub/src/workflows/package-name.ts index 42dbde4f..8169fbde 100644 --- a/apps/hub/src/workflows/package-name.ts +++ b/apps/hub/src/workflows/package-name.ts @@ -1,14 +1,4 @@ -import type { WriterDocumentKind } from "@corbits-builder/schemas/artifact"; - // The npm name each role's workflow package publishes under. A director the package ships is named // under it, since Interchange resolves a director id by its package-name prefix. export const nativePackageName = (role: string) => `@corbits-builder/${role}-native`; - -// Each writer package saves one document kind; entries/ holds one entry per name. -export const WRITER_PACKAGES = { - constraints: "writer-constraints", - approaches: "writer-approaches", - product_requirements: "writer-product-requirements", - build_plan: "writer-build-plan", -} as const satisfies Record; diff --git a/apps/hub/src/workflows/project-scope.test.ts b/apps/hub/src/workflows/project-scope.test.ts new file mode 100644 index 00000000..75c66eb3 --- /dev/null +++ b/apps/hub/src/workflows/project-scope.test.ts @@ -0,0 +1,51 @@ +import { expect, test } from "bun:test"; +import { type } from "arktype"; +import { + WORKFLOW_PACKAGES, + type InstalledWorkflows, +} from "@corbits-builder/schemas/installed-workflows"; +import { NativeSpecialistDeployment } from "@corbits-builder/schemas/native-specialist"; +import { scopeWorkflows } from "./project-scope.ts"; + +const installed = { + schemaVersion: 2, + ...Object.fromEntries( + Object.keys(WORKFLOW_PACKAGES).map((name) => [ + name, + { + source: { + kind: "asset", + assetId: "asset", + package: { format: "tarball" }, + }, + entry: "dist/workflow.mjs", + pin: `@corbits-builder/${name}-native@1.0.0-sha-new`, + sourceOfferingIds: ["offering"], + defaultSourceOfferingId: "offering", + }, + ]), + ), +} as InstalledWorkflows; + +// The scope workflows of the last release, as a bundle frozen on it parses them. +const ReleasedScopeWorkflows = type({ + "+": "reject", + writer: NativeSpecialistDeployment, + reviewer: NativeSpecialistDeployment, + coder: NativeSpecialistDeployment, + designer: NativeSpecialistDeployment, + "presenter?": NativeSpecialistDeployment, +}).onDeepUndeclaredKey("ignore"); + +test("a specialist frozen on the last release still reads its scope", () => { + for (const role of [ + "writer", + "reviewer", + "coder", + "designer", + "presenter", + ] as const) + expect( + ReleasedScopeWorkflows(scopeWorkflows(installed, role, false)), + ).not.toBeInstanceOf(type.errors); +}); diff --git a/apps/hub/src/workflows/project-scope.ts b/apps/hub/src/workflows/project-scope.ts index 7ed53d54..2c498fba 100644 --- a/apps/hub/src/workflows/project-scope.ts +++ b/apps/hub/src/workflows/project-scope.ts @@ -2,6 +2,7 @@ import { HUB_CREDENTIAL_HANDLE } from "@intx/types/credential-delivery"; import type { RuntimeCapabilities } from "@intx/types/runtime-capabilities"; import { WorkflowProjectScope, + type InstalledWorkflows, type WorkflowRole, } from "@corbits-builder/schemas/installed-workflows"; import { getLogger } from "@intx/log"; @@ -13,6 +14,32 @@ const SAFE_CODE = /^[a-z0-9_:]+$/; // A deployed agent keeps the bundle it started with; a field a newer hub adds must not refuse its scope. const HubProjectScope = WorkflowProjectScope.onDeepUndeclaredKey("ignore"); +/** The packages a run of the role may deploy or read, the coder resolved for the project. */ +export function scopeWorkflows( + installed: InstalledWorkflows, + role: WorkflowRole, + installsAnyPackage: boolean, +): WorkflowProjectScope["workflows"] { + return { + "writer-constraints": installed["writer-constraints"], + "writer-approaches": installed["writer-approaches"], + "writer-product-requirements": installed["writer-product-requirements"], + "writer-build-plan": installed["writer-build-plan"], + // A bundle frozen before writers were split requires writer to read its scope. Only a + // coordinator deploys from it, and an old one would start the constraints writer for any + // kind, so a coordinator goes without until it is upgraded. + ...(role === "coordinator" + ? {} + : { writer: installed["writer-constraints"] }), + reviewer: installed.reviewer, + coder: installsAnyPackage + ? installed["coder-any-package"] + : installed.coder, + designer: installed.designer, + presenter: installed.presenter, + }; +} + // The model sees only the original code (never a refusal body or arbitrary message); the cause stays in the hub log. export function projectScopeFailure(callId: string, error: unknown) { log.error( diff --git a/apps/hub/src/workflows/writer-package.ts b/apps/hub/src/workflows/writer-package.ts index c575e9c1..27755ec8 100644 --- a/apps/hub/src/workflows/writer-package.ts +++ b/apps/hub/src/workflows/writer-package.ts @@ -6,7 +6,8 @@ import { createSkillTools } from "../skills/tools.ts"; import { roleInference } from "./role-inference.ts"; import { compactingDirector } from "./director.ts"; import { dropDegenerateTurn, trimHistory } from "./compaction.ts"; -import { nativePackageName, WRITER_PACKAGES } from "./package-name.ts"; +import { WRITER_PACKAGES } from "@corbits-builder/schemas/installed-workflows"; +import { nativePackageName } from "./package-name.ts"; /** The writer's workflow package for one document kind, whose write_artifact saves only that kind. */ export function writerPackage(kind: WriterDocumentKind) { diff --git a/packages/schemas/src/installed-workflows.ts b/packages/schemas/src/installed-workflows.ts index fcd84e66..0f890c29 100644 --- a/packages/schemas/src/installed-workflows.ts +++ b/packages/schemas/src/installed-workflows.ts @@ -1,50 +1,68 @@ import { type } from "arktype"; import { NativeSpecialistDeployment } from "./native-specialist.ts"; import { ProjectId } from "./project.ts"; +import type { WriterDocumentKind } from "./artifact.ts"; import { WorkflowRole } from "./roles.ts"; export { SpawnedRole, WorkflowRole } from "./roles.ts"; export const INSTALLED_WORKFLOWS_KEY = "sb.installedWorkflows"; -// One writer package per document kind, each saving only its kind. -const WriterDeployments = type({ - "+": "reject", - constraints: NativeSpecialistDeployment, - approaches: NativeSpecialistDeployment, - product_requirements: NativeSpecialistDeployment, - build_plan: NativeSpecialistDeployment, -}); + +// Every workflow package Builder installs, and the role its runs play. A role names the label, model +// and guidance its packages share; a package is what a run is deployed from. +export const WORKFLOW_PACKAGES = { + coordinator: "coordinator", + "writer-constraints": "writer", + "writer-approaches": "writer", + "writer-product-requirements": "writer", + "writer-build-plan": "writer", + reviewer: "reviewer", + coder: "coder", + // The coder whose install_packages does not ask, for a project that turned asking off. + "coder-any-package": "coder", + designer: "designer", + presenter: "presenter", +} as const satisfies Record; +export type WorkflowPackage = keyof typeof WORKFLOW_PACKAGES; + +// Each writer package saves only its own document kind. +export const WRITER_PACKAGES = { + constraints: "writer-constraints", + approaches: "writer-approaches", + product_requirements: "writer-product-requirements", + build_plan: "writer-build-plan", +} as const satisfies Record; + +const writerDeployments = { + "writer-constraints": NativeSpecialistDeployment, + "writer-approaches": NativeSpecialistDeployment, + "writer-product-requirements": NativeSpecialistDeployment, + "writer-build-plan": NativeSpecialistDeployment, +} as const satisfies Record< + (typeof WRITER_PACKAGES)[WriterDocumentKind], + unknown +>; + export const InstalledWorkflows = type({ "+": "reject", - schemaVersion: "1", + schemaVersion: "2", coordinator: NativeSpecialistDeployment, - writers: WriterDeployments, + ...writerDeployments, reviewer: NativeSpecialistDeployment, coder: NativeSpecialistDeployment, - // The coder whose install_packages does not ask, for a project that turned asking off. - coderAnyPackage: NativeSpecialistDeployment, + "coder-any-package": NativeSpecialistDeployment, designer: NativeSpecialistDeployment, presenter: NativeSpecialistDeployment, }); export type InstalledWorkflows = typeof InstalledWorkflows.infer; -// Every installed package a run of each role may run on. -const DEPLOYMENTS_BY_ROLE: Record< - WorkflowRole, - (installed: InstalledWorkflows) => NativeSpecialistDeployment[] -> = { - coordinator: (installed) => [installed.coordinator], - writer: (installed) => Object.values(installed.writers), - reviewer: (installed) => [installed.reviewer], - coder: (installed) => [installed.coder, installed.coderAnyPackage], - designer: (installed) => [installed.designer], - presenter: (installed) => [installed.presenter], -}; - +/** Every installed package a run of the role may run on. */ export function roleDeployments( installed: InstalledWorkflows, role: WorkflowRole, ) { - return DEPLOYMENTS_BY_ROLE[role](installed); + return (Object.keys(WORKFLOW_PACKAGES) as WorkflowPackage[]) + .filter((name) => WORKFLOW_PACKAGES[name] === role) + .map((name) => installed[name]); } // Returned only to an authenticated native run. Scope comes from the protected @@ -57,7 +75,7 @@ export const WorkflowProjectScope = type({ role: WorkflowRole, workflows: { "+": "reject", - writers: WriterDeployments, + ...writerDeployments, // Read only by bundles frozen before writers were split; nothing deploys from it now. "writer?": NativeSpecialistDeployment, reviewer: NativeSpecialistDeployment, From a8e4ebf86a3288a635e03086ed9980a01017cbf7 Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Wed, 4 Nov 2026 12:49:36 -0800 Subject: [PATCH 4/5] fix(hub): replace an outdated coordinator before it acts on old code --- apps/hub/src/app.ts | 44 ++--- apps/hub/src/artifacts/service.ts | 5 +- apps/hub/src/auth/agent-token-bridge.ts | 89 ++++++++- apps/hub/src/config/resume.test.ts | 26 +++ apps/hub/src/config/resume.ts | 21 +- apps/hub/src/projects/coordinator-upgrade.ts | 182 ++++++++++++++++-- .../hub/src/provisioning/workflow-services.ts | 5 + apps/hub/src/workflows/project-scope.test.ts | 18 +- apps/hub/src/workflows/project-scope.ts | 14 +- packages/schemas/src/conversation.ts | 2 + 10 files changed, 331 insertions(+), 75 deletions(-) create mode 100644 apps/hub/src/config/resume.test.ts diff --git a/apps/hub/src/app.ts b/apps/hub/src/app.ts index 99e1ba86..31107f66 100644 --- a/apps/hub/src/app.ts +++ b/apps/hub/src/app.ts @@ -360,18 +360,6 @@ export async function createHubRuntime( coordinatorStartTimeoutMs, workflows.services.workflowLifecycleService.requestCancellation, ); - const coordinatorUpgrades = createCoordinatorUpgrades({ - db: handle.db, - runs: createWorkflowRunReader(agentRepoStore.repoStore), - installed: (actor, projectId) => - installForTenant( - { db: handle.db, assets: assetService }, - actor, - projectId, - ), - startup: coordinatorStartup, - cancel: workflows.services.workflowLifecycleService.requestCancellation, - }); const sidecarWebSocket = createSidecarWebSocket(sidecarRouter); const agentActivity = createAgentActivity(sidecarRouter.events); const getSession: GetSession = async (headers) => { @@ -388,6 +376,24 @@ export async function createHubRuntime( const sessionCookieName = (await auth.$context).authCookies.sessionToken .name; const app = new Hono(); + const hubMail = createHubMail( + handle.db, + async (request) => app.fetch(request), + env.hubApiOrigin, + ); + const coordinatorUpgrades = createCoordinatorUpgrades({ + db: handle.db, + runs: createWorkflowRunReader(agentRepoStore.repoStore), + installed: (actor, projectId) => + installForTenant( + { db: handle.db, assets: assetService }, + actor, + projectId, + ), + startup: coordinatorStartup, + cancel: workflows.services.workflowLifecycleService.requestCancellation, + mail: hubMail, + }); const liveAppProxy = createLiveAppProxy({ hostname: env.hostname, ports: env.liveAppPorts, @@ -419,16 +425,15 @@ export async function createHubRuntime( app.use(liveAppProxy.refuseAppOrigins); app.use(createHubContextMiddleware({ getSession })); app.use(gitTokens.middleware); - const coordinatorTokens = createAgentTokenBridge( - handle.db, - runTokenStore, - (actor, projectId) => + const coordinatorTokens = createAgentTokenBridge(handle.db, runTokenStore, { + install: (actor, projectId) => installForTenant( { db: handle.db, assets: assetService }, actor, projectId, ), - ); + handoff: coordinatorUpgrades.handoff, + }); app.use("/api/tenants/:tenantId/*", coordinatorTokens.middleware); app.use("/api/workflow-project-scope", coordinatorTokens.middleware); app.use( @@ -627,11 +632,6 @@ export async function createHubRuntime( 500, ); }); - const hubMail = createHubMail( - handle.db, - async (request) => app.fetch(request), - env.hubApiOrigin, - ); const projectSchedulers = [ createReconciliationScheduler({ name: "Specialist end notice", diff --git a/apps/hub/src/artifacts/service.ts b/apps/hub/src/artifacts/service.ts index bd78ae16..94544eda 100644 --- a/apps/hub/src/artifacts/service.ts +++ b/apps/hub/src/artifacts/service.ts @@ -1,7 +1,7 @@ import { stat } from "node:fs/promises"; import { dirname, join } from "node:path"; import { type } from "arktype"; -import { and, desc, eq, inArray, ne, sql } from "drizzle-orm"; +import { and, desc, eq, inArray, ne, notInArray, sql } from "drizzle-orm"; import { PostgresJsDatabase } from "drizzle-orm/postgres-js"; import { createArtifact, @@ -975,7 +975,8 @@ export function createArtifactService( .where( and( eq(approval.tenantId, actor.tenantId), - ne(approval.status, "pending"), + // An expired ask is one a coordinator upgrade withdrew; its successor asks again. + notInArray(approval.status, ["pending", "expired"]), ), ) .orderBy(approval.createdAt); diff --git a/apps/hub/src/auth/agent-token-bridge.ts b/apps/hub/src/auth/agent-token-bridge.ts index b30aa7a6..16aebf3c 100644 --- a/apps/hub/src/auth/agent-token-bridge.ts +++ b/apps/hub/src/auth/agent-token-bridge.ts @@ -2,6 +2,7 @@ import { type } from "arktype"; import { InstalledWorkflows, INSTALLED_WORKFLOWS_KEY, + WRITER_PACKAGES, type WorkflowRole, } from "@corbits-builder/schemas/installed-workflows"; import { projectCoordinator } from "../db/schema.ts"; @@ -23,7 +24,7 @@ import { type AppEnv, } from "@intx/hub-api"; import { and, eq } from "drizzle-orm"; -import type { MiddlewareHandler } from "hono"; +import type { Context, MiddlewareHandler } from "hono"; import { getLogger } from "@intx/log"; import type { Actor } from "../lib/actor.ts"; import { createCoordinatorRegistry } from "./coordinator-registry.ts"; @@ -31,6 +32,9 @@ import { projectResource } from "../projects/errors.ts"; import { spawnedWorkflowRole } from "../workflows/spawn-role.ts"; import { buildPackageLayout } from "../builds/package-layout.ts"; import { scopeWorkflows } from "../workflows/project-scope.ts"; +import { nativePackageName } from "../workflows/package-name.ts"; +import { readJson } from "../lib/json.ts"; +import { coordinatorUpgrading } from "@corbits-builder/schemas/conversation"; import { installsAnyPackage, opensNetwork } from "../config/preferences.ts"; import { loadSettings } from "../settings/policy.ts"; @@ -113,6 +117,63 @@ function isNativeWorkflowDeploy(method: string, path: string): boolean { ); } +// A deploy of any writer package. An outdated coordinator names the constraints writer whatever +// document it means, so that deploy is refused while a handoff waits on busy specialists. +async function startsWriter(request: Request) { + const body = await readJson(request.clone()); + const pin = + typeof body === "object" && body !== null && "pin" in body + ? body.pin + : undefined; + return ( + typeof pin === "string" && + Object.values(WRITER_PACKAGES).some((name) => + pin.startsWith(nativePackageName(name) + "@"), + ) + ); +} + +// The replaced coordinator is stopped within about a second. Answering would hand its model a +// result to act on and talk about while it still runs, so the call stays open until the stop closes +// it; a stop that never comes ends in the same refusal a busy handoff gives. +const HELD_DEPLOY_MS = 30_000; +async function heldUntilStopped(c: Context) { + const closed = c.req.raw.signal; + await new Promise((resolve) => { + const timer = setTimeout(resolve, HELD_DEPLOY_MS); + closed.addEventListener( + "abort", + () => { + clearTimeout(timer); + resolve(); + }, + { once: true }, + ); + }); + return c.json({ code: coordinatorUpgrading }, 409); +} + +// Right after an approval the writer that filed the document is still finishing, so a coordinator +// replaced at once would be refused and retry in front of the person. The call waits for the +// specialists to settle and is handed off then; only a wait that outlasts the hold is refused. +const SETTLE_RECHECK_MS = 1_000; +async function settledHandoff( + c: Context, + check: () => Promise<"current" | "busy" | "handed_off">, +) { + const until = Date.now() + HELD_DEPLOY_MS; + let outcome = await check(); + while ( + outcome === "busy" && + Date.now() < until && + !c.req.raw.signal.aborted + ) { + await Bun.sleep(SETTLE_RECHECK_MS); + outcome = await check(); + } + return outcome; +} + type RunTokenLookup = { findByHash( tokenHash: Uint8Array, @@ -170,8 +231,14 @@ async function nativeRunIdentity( export function createAgentTokenBridge( db: ReturnType["db"], runTokens?: RunTokenLookup, - // Reinstalls a tenant whose stored packages predate this hub, so a run is never refused for it. - install?: (actor: Actor, projectId: string) => Promise, + upgrades?: { + // Reinstalls a tenant whose stored packages predate this hub, so a run is never refused for it. + install: (actor: Actor, projectId: string) => Promise; + handoff: ( + tenantId: string, + runId: string, + ) => Promise<"current" | "busy" | "handed_off">; + }, ) { const registry = createCoordinatorRegistry(db); const authenticated = new WeakSet(); @@ -236,12 +303,21 @@ export function createAgentTokenBridge( runTokens === undefined ? null : await nativeRunIdentity(db, runTokens, bearer); - if (deployer?.runId && deployer.tenantId === c.req.param("tenantId")) + if (deployer?.runId && deployer.tenantId === c.req.param("tenantId")) { + if (upgrades) { + const outcome = await settledHandoff(c, () => + upgrades.handoff(deployer.tenantId, deployer.runId), + ); + if (outcome === "handed_off") return heldUntilStopped(c); + if (outcome === "busy" && (await startsWriter(c.req.raw))) + return c.json({ code: coordinatorUpgrading }, 409); + } await mirrorDelegatableDeployerGrants(db, { tenantId: deployer.tenantId, runId: deployer.runId, principalId: deployer.id, }); + } } if (!nativeBearer) recordRefusal("token_unverified"); return next(); @@ -287,9 +363,9 @@ export function createAgentTokenBridge( const stored = InstalledWorkflows(config?.[INSTALLED_WORKFLOWS_KEY]); const installed = stored instanceof type.errors - ? install && + ? upgrades && binding.run.principalId && - (await install( + (await upgrades.install( { tenantId: row.id, principalId: binding.run.principalId }, binding.projectId, )) @@ -307,7 +383,6 @@ export function createAgentTokenBridge( role, workflows: scopeWorkflows( installed, - role, installsAnyPackage(preferences.value), ), openNetwork: opensNetwork(preferences.value), diff --git a/apps/hub/src/config/resume.test.ts b/apps/hub/src/config/resume.test.ts new file mode 100644 index 00000000..8f651d0a --- /dev/null +++ b/apps/hub/src/config/resume.test.ts @@ -0,0 +1,26 @@ +import { expect, test } from "bun:test"; +import { + ASKING_HANDOFF_OPENING, + HANDOFF_OPENING, + resumeContext, + resumeMessage, + withoutResumeContext, +} from "./resume.ts"; + +// The hub mails the handoff note as the person so the replacement coordinator acts on it; shown in +// the conversation, it would read as something the person said. +test("the hub's handoff note never shows as the person's message", () => { + const context = resumeContext({ + turns: [{ role: "person", text: "Build a weather app", at: 1 }], + awaiting: [], + withApp: false, + }); + expect(withoutResumeContext(HANDOFF_OPENING)).toBeUndefined(); + expect(withoutResumeContext(ASKING_HANDOFF_OPENING)).toBeUndefined(); + expect( + withoutResumeContext(resumeMessage(context, HANDOFF_OPENING)), + ).toBeUndefined(); + expect(withoutResumeContext(resumeMessage(context, "Go ahead"))).toBe( + "Go ahead", + ); +}); diff --git a/apps/hub/src/config/resume.ts b/apps/hub/src/config/resume.ts index e2a36a43..e60a8cf5 100644 --- a/apps/hub/src/config/resume.ts +++ b/apps/hub/src/config/resume.ts @@ -12,6 +12,13 @@ const RESUME_CONTEXT_END = "\n\nThe person's new message:\n\n"; // Hubs before the held resume context mailed this opening on the person's behalf. const LEGACY_RESUME_OPENING = "I'm picking this project back up."; +// Sent by the hub, never shown to the person, when a coordinator on an outdated version was replaced +// as it started a writer: its writer did not start, and the person is not waiting to be asked. +const HANDOFF_LEAD = + "Hub note, not from the person: Builder moved this project to a new version"; +export const HANDOFF_OPENING = `${HANDOFF_LEAD} as you were starting a specialist, so it did not start. Check which documents are filed and approved, start the specialist the work needs next, and carry on without mentioning the move.`; +export const ASKING_HANDOFF_OPENING = `${HANDOFF_LEAD} as you were asking the person to approve something, so they never saw the question. Check which documents are filed and approved, ask for the approval the work needs next, and carry on without mentioning the move.`; + const TURN_CHARS = 2_000; const CONTEXT_CHARS = 16_000; @@ -74,9 +81,13 @@ export function approvalDecisionLine({ /** A person-sent turn without the held resume context, or undefined for a hub-written opening. */ export function withoutResumeContext(text: string): string | undefined { if (text.startsWith(LEGACY_RESUME_OPENING)) return undefined; - if (!text.startsWith(RESUME_CONTEXT_LEAD)) return text; - const end = text.indexOf(RESUME_CONTEXT_END); - if (end !== -1) return text.slice(end + RESUME_CONTEXT_END.length); - // Delivered text is trimmed, so a blank new message leaves the marker without its blank line. - return text.endsWith(RESUME_CONTEXT_END.trimEnd()) ? "" : text; + let message = text; + if (text.startsWith(RESUME_CONTEXT_LEAD)) { + const end = text.indexOf(RESUME_CONTEXT_END); + // Delivered text is trimmed, so a blank new message leaves the marker without its blank line. + if (end === -1) + return text.endsWith(RESUME_CONTEXT_END.trimEnd()) ? "" : text; + message = text.slice(end + RESUME_CONTEXT_END.length); + } + return message.startsWith(HANDOFF_LEAD) ? undefined : message; } diff --git a/apps/hub/src/projects/coordinator-upgrade.ts b/apps/hub/src/projects/coordinator-upgrade.ts index 7274dd14..e26baeef 100644 --- a/apps/hub/src/projects/coordinator-upgrade.ts +++ b/apps/hub/src/projects/coordinator-upgrade.ts @@ -1,6 +1,6 @@ -import { and, desc, eq, inArray } from "drizzle-orm"; +import { and, desc, eq, inArray, ne } from "drizzle-orm"; import { type } from "arktype"; -import type { DB } from "@intx/db"; +import { createApprovalStore, type DB } from "@intx/db"; import { approval, workflowRun, @@ -21,6 +21,8 @@ import { import type { ProjectUpdateReceipt } from "@corbits-builder/schemas/project"; import { spawnedWorkflowRole } from "../workflows/spawn-role.ts"; import { MODEL_CHOICE_REFUSALS } from "../config/models.ts"; +import { ASKING_HANDOFF_OPENING, HANDOFF_OPENING } from "../config/resume.ts"; +import type { HubMail } from "../mail/hub-mail.ts"; import { project, projectCoordinator } from "../db/schema.ts"; import type { Actor } from "../lib/actor.ts"; import { settle } from "../lib/serial-queue.ts"; @@ -85,10 +87,12 @@ export function createCoordinatorUpgrades(deps: { release(actor: Actor, projectId: string): Promise; }; cancel: (tenantId: string, runId: string, reason: string) => Promise; + mail: HubMail; }) { const { db } = deps; + const approvals = createApprovalStore(db); - async function parkedOnInput(run: { runId: string; address: string }) { + async function parkOf(run: { runId: string; address: string }) { const events = await deps.runs.readRunEvents( workflowRunRepoIdForAddress(run.address), WORKFLOW_RUN_REF, @@ -98,7 +102,11 @@ export function createCoordinatorUpgrades(deps: { (event) => event.type === "SignalAwaited" || event.type === "SignalReceived", ); - return last?.type === "SignalAwaited" && last.body["parkKind"] === "input"; + return last?.type === "SignalAwaited" ? last.body["parkKind"] : undefined; + } + + async function parkedOnInput(run: { runId: string; address: string }) { + return (await parkOf(run)) === "input"; } // The specialists to close with the coordinator, or undefined while one is still at work. A designer @@ -140,40 +148,66 @@ export function createCoordinatorUpgrades(deps: { return parked; } - // Parked on the person's input with every delivery to it settled. - async function coordinatorIdle(run: { runId: string; address: string }) { - if (!(await parkedOnInput(run))) return false; + // Mail or a signal the run has not yet taken; a run stopped now would lose it. + async function undelivered(runId: string) { const [delivery] = await db .select({ id: workflowRunDispatch.id }) .from(workflowRunDispatch) .where( and( - eq(workflowRunDispatch.anchorRunId, run.runId), + eq(workflowRunDispatch.anchorRunId, runId), inArray(workflowRunDispatch.status, ["pending", "acknowledged"]), ), ) .limit(1); - return delivery === undefined; + return delivery !== undefined; + } + + // Parked on the person's input with every delivery to it settled. + async function coordinatorIdle(run: { runId: string; address: string }) { + return (await parkedOnInput(run)) && !(await undelivered(run.runId)); } - async function approvalPending(tenantId: string) { + async function approvalPending(tenantId: string, besides?: string) { const [asked] = await db .select({ id: approval.id }) .from(approval) .where( - and(eq(approval.tenantId, tenantId), eq(approval.status, "pending")), + and( + eq(approval.tenantId, tenantId), + eq(approval.status, "pending"), + besides === undefined ? undefined : ne(approval.runId, besides), + ), ) .limit(1); return asked !== undefined; } + // The coordinator's own undecided approval, which it waits on. + async function ownApproval(runId: string) { + const [asked] = await db + .select({ correlationId: approval.correlationId }) + .from(approval) + .where(and(eq(approval.runId, runId), eq(approval.status, "pending"))) + .limit(1); + return asked?.correlationId; + } + + // A coordinator waiting on its own approval is replaced too: approved, the outdated run would act + // on it with old code, so the person would be asked again by its successor. async function atSafePoint( actor: Actor, run: { runId: string; address: string }, ) { - if (!(await coordinatorIdle(run))) return undefined; - if (await approvalPending(actor.tenantId)) return undefined; - return settledSpecialists(actor.tenantId); + if (await undelivered(run.runId)) return undefined; + const park = await parkOf(run); + const asked = + park === "approval" ? await ownApproval(run.runId) : undefined; + if (park !== "input" && asked === undefined) return undefined; + if (await approvalPending(actor.tenantId, asked && run.runId)) + return undefined; + const parked = await settledSpecialists(actor.tenantId); + return parked && { parked, asked }; } async function start(actor: Actor, projectId: string) { @@ -187,8 +221,9 @@ export function createCoordinatorUpgrades(deps: { eq(projectCoordinator.active, true), ), ); - if (!bound) return; + if (!bound) return undefined; log.info("Coordinator upgraded", { projectId, runId: bound.runId }); + return bound.runId; } // The project's active coordinator, or the latest one when none is active. @@ -245,16 +280,100 @@ export function createCoordinatorUpgrades(deps: { return; const current = (await deps.installed(actor, projectId)).coordinator.pin; if (pinOf(latest.bundle) === current) return; - const parked = await atSafePoint(actor, { + const safe = await atSafePoint(actor, { runId: latest.runId, address: latest.address, }); - if (!parked) return; + if (!safe) return; await deps.startup.release(actor, projectId); - for (const runId of parked) + for (const runId of safe.parked) await deps.cancel(tenantId, runId, COORDINATOR_UPGRADED); await deps.cancel(tenantId, latest.runId, COORDINATOR_UPGRADED); - await start(actor, projectId); + if (safe.asked === undefined) { + await start(actor, projectId); + return; + } + // Interchange resolves an approval only through its decision, which a stopping run cannot take, + // so the question is closed here or it would wait on the person forever. + await approvals.resolve(safe.asked, { + status: "expired", + resolvedAt: new Date(), + }); + const next = await start(actor, projectId); + if (next) + await deps.mail({ + tenantId, + runId: next, + principalId: actor.principalId, + content: ASKING_HANDOFF_OPENING, + }); + } + + /** + * Replaces an outdated coordinator the moment it starts a specialist, before the deploy lands. A + * coordinator from before writers were split reads one writer entry, the constraints writer, and + * would start it for whatever document it means; the replacement on the current bundle picks the + * writer from the documents. It is told at once, so the person sees the work carry on rather than + * wait for their next message. Interchange drops mail to a stopping run, so a specialist still at + * work or a delivery the coordinator has not taken leaves it to the safe-point pass instead. + */ + async function handoff( + projectId: string, + tenantId: string, + runId: string, + ): Promise<"current" | "busy" | "handed_off"> { + const latest = await latestCoordinator(projectId, tenantId); + if (!latest || latest.runId !== runId) return "handed_off"; + if ( + !latest.active || + latest.cancellationRequestedAt !== null || + !["deployed", "running"].includes(latest.status) + ) + return "handed_off"; + const [parent] = await db + .select({ status: project.status }) + .from(project) + .where(eq(project.id, projectId)); + if (parent?.status !== "active") return "handed_off"; + const actor = { tenantId, principalId: latest.principalId }; + let current: string; + try { + current = (await deps.installed(actor, projectId)).coordinator.pin; + } catch (error) { + // The pass and the person's next open meet the same refusal; this deploy goes ahead as it did + // before handoffs existed. + log.warn("Coordinator handoff check failed", { + projectId, + ...requestErrorMetadata(error), + }); + return "current"; + } + if (pinOf(latest.bundle) === current) return "current"; + if (await undelivered(runId)) return "busy"; + const parked = await settledSpecialists(tenantId); + if (!parked) return "busy"; + await deps.startup.release(actor, projectId); + for (const id of parked) + await deps.cancel(tenantId, id, COORDINATOR_UPGRADED); + await deps.cancel(tenantId, runId, COORDINATOR_UPGRADED); + // The old run is stopping either way, so its deploy is held whatever happens next; the pass + // starts a successor that failed to start here, as it does after any replacement. + try { + const next = await start(actor, projectId); + if (next) + await deps.mail({ + tenantId, + runId: next, + principalId: actor.principalId, + content: HANDOFF_OPENING, + }); + } catch (error) { + log.error("Coordinator handoff start failed", { + projectId, + ...requestErrorMetadata(error), + }); + } + return "handed_off"; } /** @@ -383,8 +502,33 @@ export function createCoordinatorUpgrades(deps: { return work; } + // A deploy from a project's coordinator, checked under the project's upgrade lock so a pass, an + // update or an archive never replaces the same coordinator at the same time. A deploy from a run + // that is no coordinator, a specialist, goes ahead. + async function handoffOnce(tenantId: string, runId: string) { + const [bound] = await db + .select({ projectId: projectCoordinator.projectId }) + .from(projectCoordinator) + .where(eq(projectCoordinator.runId, runId)); + if (!bound) return "current"; + const { projectId } = bound; + const pending = inflight.get(projectId); + async function handoffAfterPending() { + try { + await settle(pending?.work); + return await handoff(projectId, tenantId, runId); + } finally { + if (inflight.get(projectId)?.work === work) inflight.delete(projectId); + } + } + const work = handoffAfterPending(); + inflight.set(projectId, { opened: true, work }); + return work; + } + return { reconcile, + handoff: handoffOnce, open: (actor: Actor, projectId: string) => upgradeOnce(projectId, actor.tenantId, actor), update: updateOnce, diff --git a/apps/hub/src/provisioning/workflow-services.ts b/apps/hub/src/provisioning/workflow-services.ts index a90b50cb..55423c85 100644 --- a/apps/hub/src/provisioning/workflow-services.ts +++ b/apps/hub/src/provisioning/workflow-services.ts @@ -240,6 +240,11 @@ export async function createWorkflowServices(options: { historyReceives, sendControl: (target, command, timeout) => router.sendWorkflowControl(target, command, timeout), + // A graceful cancel ends only the turn of a single-step agent such as the coordinator; its model + // keeps answering tool results until the run is stopped. No Builder workflow does work on + // cancel, so every cancel here (replacement, archive, lifetime, start timeout) stops the run on + // the next sweep, and a replaced coordinator cannot act or talk after it. + cancelGraceMs: 0, }); const projection = createWorkflowDispatchProjection({ db, diff --git a/apps/hub/src/workflows/project-scope.test.ts b/apps/hub/src/workflows/project-scope.test.ts index 75c66eb3..c3477177 100644 --- a/apps/hub/src/workflows/project-scope.test.ts +++ b/apps/hub/src/workflows/project-scope.test.ts @@ -27,7 +27,8 @@ const installed = { ), } as InstalledWorkflows; -// The scope workflows of the last release, as a bundle frozen on it parses them. +// The scope workflows as a bundle frozen on the last release parses them. A coordinator that could +// not read its scope failed the tool that carried the person's approval and asked for it again. const ReleasedScopeWorkflows = type({ "+": "reject", writer: NativeSpecialistDeployment, @@ -37,15 +38,8 @@ const ReleasedScopeWorkflows = type({ "presenter?": NativeSpecialistDeployment, }).onDeepUndeclaredKey("ignore"); -test("a specialist frozen on the last release still reads its scope", () => { - for (const role of [ - "writer", - "reviewer", - "coder", - "designer", - "presenter", - ] as const) - expect( - ReleasedScopeWorkflows(scopeWorkflows(installed, role, false)), - ).not.toBeInstanceOf(type.errors); +test("a run frozen on the last release still reads its scope", () => { + expect( + ReleasedScopeWorkflows(scopeWorkflows(installed, false)), + ).not.toBeInstanceOf(type.errors); }); diff --git a/apps/hub/src/workflows/project-scope.ts b/apps/hub/src/workflows/project-scope.ts index 2c498fba..0493435b 100644 --- a/apps/hub/src/workflows/project-scope.ts +++ b/apps/hub/src/workflows/project-scope.ts @@ -14,10 +14,9 @@ const SAFE_CODE = /^[a-z0-9_:]+$/; // A deployed agent keeps the bundle it started with; a field a newer hub adds must not refuse its scope. const HubProjectScope = WorkflowProjectScope.onDeepUndeclaredKey("ignore"); -/** The packages a run of the role may deploy or read, the coder resolved for the project. */ +/** The packages any run in the project may deploy or read, the coder resolved for the project. */ export function scopeWorkflows( installed: InstalledWorkflows, - role: WorkflowRole, installsAnyPackage: boolean, ): WorkflowProjectScope["workflows"] { return { @@ -25,12 +24,11 @@ export function scopeWorkflows( "writer-approaches": installed["writer-approaches"], "writer-product-requirements": installed["writer-product-requirements"], "writer-build-plan": installed["writer-build-plan"], - // A bundle frozen before writers were split requires writer to read its scope. Only a - // coordinator deploys from it, and an old one would start the constraints writer for any - // kind, so a coordinator goes without until it is upgraded. - ...(role === "coordinator" - ? {} - : { writer: installed["writer-constraints"] }), + // A bundle frozen before writers were split cannot read its scope without writer, and a + // coordinator waiting on the person must still finish their approval. Starting a writer from it + // would start the constraints writer for any document, so the hub hands such a coordinator off + // before that deploy lands (coordinator-upgrade handoff). + writer: installed["writer-constraints"], reviewer: installed.reviewer, coder: installsAnyPackage ? installed["coder-any-package"] diff --git a/packages/schemas/src/conversation.ts b/packages/schemas/src/conversation.ts index a8be5e30..0ca75287 100644 --- a/packages/schemas/src/conversation.ts +++ b/packages/schemas/src/conversation.ts @@ -56,6 +56,8 @@ const conversationTextTurn = type({ at: "number", }); export const maxConversationTurns = 2000; +/** The hub's refusal of a deploy from a coordinator it is replacing; the replacement redoes it. */ +export const coordinatorUpgrading = "coordinator_upgrading"; const turnList = conversationTextTurn .array() .atMostLength(maxConversationTurns); From 952ac675d9d851c1ad0ab502f645759d57309e47 Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Wed, 4 Nov 2026 12:49:38 -0800 Subject: [PATCH 5/5] fix(web): keep the conversation on screen between coordinators --- apps/hub/src/app.ts | 3 + apps/hub/src/mail/conversation-routes.ts | 123 +++++++++++++----- apps/web/src/api/conversation.ts | 38 ++---- apps/web/src/api/decisions.ts | 2 +- .../web/src/pages/workspace/dispatch-card.tsx | 7 + apps/web/src/pages/workspace/index.tsx | 33 ++++- apps/web/src/pages/workspace/load-room.ts | 5 +- packages/schemas/src/conversation.ts | 2 +- 8 files changed, 148 insertions(+), 65 deletions(-) diff --git a/apps/hub/src/app.ts b/apps/hub/src/app.ts index 31107f66..7ec775cc 100644 --- a/apps/hub/src/app.ts +++ b/apps/hub/src/app.ts @@ -513,6 +513,9 @@ export async function createHubRuntime( createConversationRoutes(handle.db, (request) => app.fetch(request), { repoStore: agentRepoStore.repoStore, mailDirectory: reportMailDirectory, + ...(coordinatorUpgrade.enabled + ? { upgrade: coordinatorUpgrades.open } + : {}), }), ); app.route( diff --git a/apps/hub/src/mail/conversation-routes.ts b/apps/hub/src/mail/conversation-routes.ts index 1991aab3..6e86bae6 100644 --- a/apps/hub/src/mail/conversation-routes.ts +++ b/apps/hub/src/mail/conversation-routes.ts @@ -26,6 +26,7 @@ import { type CoordinatorRepoDirSource, } from "./coordinator-question.ts"; import { isRecord, projectRunTurns } from "./conversation-turns.ts"; +import type { Actor } from "../lib/actor.ts"; const log = getLogger(["corbits-builder", "hub", "conversation"]); @@ -38,8 +39,12 @@ export function createConversationRoutes( repoStore?: CoordinatorRepoDirSource; /** The retained native mail store the hub archives workflow mail into. */ mailDirectory?: string; + /** Replaces an outdated coordinator, which may be waiting on an approval the person is reading. */ + upgrade?: (actor: Actor, projectId: string) => Promise; }, ) { + // The approval park each coordinator was last checked at, so a person reading it checks it once. + const checked = new Map(); const app = new Hono(); app.get( "/:projectId/conversation/events", @@ -54,7 +59,7 @@ export function createConversationRoutes( operationId: "readProjectConversationEvents", tags: ["Conversation"], description: - "Read the current registered coordinator's committed native runtime events. These events are not a rendered conversation transcript by themselves. When present, turns are the person's thread across every coordinator run of the project, in timestamp order: person and coordinator text from the committed durable conversation mirror, plus participant mail from other workflow runs on a retained mail thread that includes a person. threads are the retained native mail threads among workflow runs only, keyed by References root. A still-open input park also carries the coordinator's asking prose when the run's committed durable conversation mirror provides it (body.question, sourced from the coordinator's turn); parks without real prose stay verbatim. Native workflow read authority is required in addition to project read.", + "Read the current registered coordinator's committed native runtime events. While no coordinator is registered (one is starting or replacing another), runId is null, events is empty and turns still carry the conversation of the project's earlier coordinator runs. These events are not a rendered conversation transcript by themselves. When present, turns are the person's thread across every coordinator run of the project, in timestamp order: person and coordinator text from the committed durable conversation mirror, plus participant mail from other workflow runs on a retained mail thread that includes a person. threads are the retained native mail threads among workflow runs only, keyed by References root. A still-open input park also carries the coordinator's asking prose when the run's committed durable conversation mirror provides it (body.question, sourced from the coordinator's turn); parks without real prose stay verbatim. Native workflow read authority is required in addition to project read.", security: [{ sessionCookie: [] }, { deviceBearer: [] }], parameters: ["tenantId", "projectId"].map((name) => ({ name, @@ -75,7 +80,6 @@ export function createConversationRoutes( description: "User, project or native workflow authority refused", }, 404: { description: "Project or native run unavailable" }, - 409: { description: "Coordinator not registered or unavailable" }, 503: { description: "Native event history unavailable" }, }, }), @@ -119,31 +123,79 @@ export function createConversationRoutes( ), ) .orderBy(asc(projectCoordinator.createdAt)); - const run = runs.find((candidate) => candidate.active); - if (!run || run.run.anchorRunId !== run.run.id || !run.run.address) - return null; + const found = runs.find((candidate) => candidate.active); + const run = + found && found.run.anchorRunId === found.run.id && found.run.address + ? { runId: found.run.id, address: found.run.address } + : undefined; // A resumed project runs on a new coordinator; earlier runs keep // the conversation that came before it. const prior = runs.flatMap((candidate) => - candidate.active || !candidate.run.address + candidate === found || !candidate.run.address ? [] : [{ runId: candidate.run.id, address: candidate.run.address }], ); - return { runId: run.run.id, address: run.run.address, prior }; + return { run, prior }; }); - if (!selected) throw new ProjectError(409, "coordinator_unavailable"); + const repoStore = options?.repoStore; + const mailDirectory = options?.mailDirectory; + const read = ( + store: CoordinatorRepoDirSource, + run: { runId: string; address: string }, + ) => + projectRunTurns({ + db, + tenantId, + repoStore: store, + mailDirectory, + ...run, + }); + const priorTurns = async (store: CoordinatorRepoDirSource) => + ( + await Promise.all(selected.prior.map((run) => read(store, run))) + ).flatMap((prior) => prior.turns); + // Between one coordinator and the next (an upgrade hands off, a reopened project starts + // one) the person still reads what was said; only the live run's events wait for a bind. + if (selected.run === undefined) { + // Without the mirror store there are no turns to read, as for a bound coordinator. + if (repoStore === undefined) + return c.json({ runId: null, events: [] }); + let turns; + try { + turns = (await priorTurns(repoStore)).sort( + (left, right) => left.at - right.at, + ); + } catch (error) { + log.error( + "Conversation projection failed", + requestErrorMetadata(error), + ); + throw new ProjectError( + 503, + "conversation_projection_failed", + {}, + { cause: error }, + ); + } + return c.json({ + runId: null, + events: [], + turns: turns.slice(-maxConversationTurns), + }); + } + const current = selected.run; const target = new URL(c.req.url); target.pathname = "/api/tenants/" + encodeURIComponent(tenantId) + "/workflows/runs/" + - encodeURIComponent(selected.runId) + + encodeURIComponent(current.runId) + "/events"; // Product query parameters must not steer the native target. target.search = ""; const response = await dispatch(new Request(target.href, c.req.raw)); response.headers.set("cache-control", "no-store"); - if (!response.ok || options?.repoStore === undefined) return response; + if (!response.ok || repoStore === undefined) return response; // The input park carries no wording; when the durable conversation // mirror provides asking prose, it rides the pending park. A missing // mirror is an empty turn list, not a failed read. @@ -156,35 +208,45 @@ export function createConversationRoutes( }); const payload = parseJson(raw); if (payload === undefined) return passThrough(raw); - const { address } = selected; + const { address } = current; const events = (payload as { events?: unknown })["events"]; if (Array.isArray(events)) { + const asking = latestPendingPark(events, "approval"); + const upgrade = options?.upgrade; + if ( + asking !== undefined && + upgrade !== undefined && + checked.get(current.runId) !== asking["correlationId"] + ) { + checked.set(current.runId, asking["correlationId"]); + // The pass would leave the question up to its interval; replacing the coordinator + // now withdraws it before the person decides it. + const withdraw = async () => { + try { + await upgrade({ tenantId, principalId: actor.id }, projectId); + } catch (error) { + log.warn( + "Coordinator upgrade on approval failed", + requestErrorMetadata(error), + ); + } + }; + void withdraw(); + } const readProse = createCoordinatorQuestionProseReader({ - repoStore: options.repoStore, + repoStore, }); - const park = latestPendingInputPark(events); + const park = latestPendingPark(events, "input"); if (park !== undefined) { const question = await readProse({ address, park }); if (question !== undefined) park["question"] = question; } } if (isRecord(payload)) { - const { repoStore, mailDirectory } = options; - const read = (run: { runId: string; address: string }) => - projectRunTurns({ - db, - tenantId, - repoStore, - mailDirectory, - ...run, - }); - const current = await read({ runId: selected.runId, address }); - const turns = [...current.turns]; - for (const run of selected.prior) - turns.push(...(await read(run)).turns); + const live = await read(repoStore, current); + const turns = [...live.turns, ...(await priorTurns(repoStore))]; turns.sort((left, right) => left.at - right.at); - if (current.threads !== undefined) - payload["threads"] = current.threads; + if (live.threads !== undefined) payload["threads"] = live.threads; payload["turns"] = turns.slice(-maxConversationTurns); } return passThrough(JSON.stringify(payload)); @@ -216,8 +278,9 @@ export function createConversationRoutes( * bodies, so wording is attributed only to the question the coordinator is * waiting on. */ -function latestPendingInputPark( +function latestPendingPark( events: unknown[], + kind: "input" | "approval", ): Record | undefined { for (let index = events.length - 1; index >= 0; index -= 1) { const event = events[index]; @@ -226,7 +289,7 @@ function latestPendingInputPark( if (type !== "SignalAwaited" || typeof body !== "object" || body === null) continue; const park = body as Record; - if (park["parkKind"] !== "input") continue; + if (park["parkKind"] !== kind) continue; const released = events .slice(index + 1) .some( diff --git a/apps/web/src/api/conversation.ts b/apps/web/src/api/conversation.ts index 7d97e248..b3f1f7f4 100644 --- a/apps/web/src/api/conversation.ts +++ b/apps/web/src/api/conversation.ts @@ -8,6 +8,7 @@ * shape and `events` derives the latest person-input question (an input park * and its releasing signal) from the events. */ +import type { ConversationEvents } from "@corbits-builder/schemas/conversation"; import { wording } from "../config.ts"; import { productClient } from "./transport.ts"; import { @@ -31,35 +32,16 @@ export type ConversationThread = { }; /** - * The committed conversation, or null while the project has no bound - * coordinator (the hub answers 409 `{ code: "coordinator_unavailable" }`). + * The committed conversation. `runId` is null while no coordinator is bound; the turns still + * carry what the project's earlier coordinators said. */ -export async function readConversationEvents(projectId: string): Promise<{ - runId: string; - events: ConversationEvent[]; - turns: ConversationTurn[]; - threads: ConversationThread[]; -} | null> { - const client = productClient(); - const tenantId = await workspaceTenantId(); - try { - const body = await client.conversationEvents(tenantId, projectId); - return { - runId: body.runId, - events: body.events, - turns: body.turns ?? [], - threads: body.threads ?? [], - }; - } catch (cause) { - const status = - cause instanceof Error && - "response" in cause && - cause.response instanceof Response - ? cause.response.status - : undefined; - if (status === 409) return null; - throw cause; - } +export async function readConversationEvents( + projectId: string, +): Promise { + return productClient().conversationEvents( + await workspaceTenantId(), + projectId, + ); } export type SendReceipt = { runId: string; address: string; messageId: string }; diff --git a/apps/web/src/api/decisions.ts b/apps/web/src/api/decisions.ts index 2bff0e12..05df3a1a 100644 --- a/apps/web/src/api/decisions.ts +++ b/apps/web/src/api/decisions.ts @@ -90,7 +90,7 @@ export async function openDecisions(onlyProjectId?: string): Promise { readConversationEvents(record.projectId), client.nativeApprovals(record.tenantId), ]); - if (conversation === null) continue; + if (conversation.runId === null) continue; const pending = approvals.filter( (approval) => approval.status === "pending", ); diff --git a/apps/web/src/pages/workspace/dispatch-card.tsx b/apps/web/src/pages/workspace/dispatch-card.tsx index 57b31725..09118dca 100644 --- a/apps/web/src/pages/workspace/dispatch-card.tsx +++ b/apps/web/src/pages/workspace/dispatch-card.tsx @@ -9,6 +9,7 @@ import { createContext } from "react"; import type { ConversationPart, PartTool } from "@corbits/react-ui"; import type { AgentActivity, ProjectArtifact } from "@corbits-builder/client"; import type { FailedRun, InFlightRun } from "@corbits-builder/schemas/artifact"; +import { coordinatorUpgrading } from "@corbits-builder/schemas/conversation"; import { wording } from "../../config.ts"; import type { CallResult } from "./live-run.ts"; import type { ApprovalOutcome, DocumentApproval } from "./transcript.tsx"; @@ -88,6 +89,12 @@ export function dispatchesOf( ...(kind === undefined ? {} : { kind }), }; const output = result === undefined ? undefined : parsed(result.content); + // A replaced coordinator's start is redone by its successor, so it is not the person's failure. + if ( + result?.isError === true && + str(result.content)?.includes(coordinatorUpgrading) === true + ) + return []; if (result?.isError === true) return [ { diff --git a/apps/web/src/pages/workspace/index.tsx b/apps/web/src/pages/workspace/index.tsx index 836e32bb..c0dcc804 100644 --- a/apps/web/src/pages/workspace/index.tsx +++ b/apps/web/src/pages/workspace/index.tsx @@ -146,6 +146,11 @@ import { useModelCatalog } from "../providers.tsx"; // The stream says when the room moved. This read covers what it cannot see, // such as a coordinator run replaced while its old stream stays quiet. const FALLBACK_RELOAD_MS = 30_000; +// No coordinator's stream announces the next one binding, so an unbound room reads again until it does. +const BIND_POLL_MS = 1_000; +// A handoff binds within seconds; past this the room starts the coordinator itself, so a start the +// hub could not make shows its refusal instead of an endless thinking mark. +const BIND_WAIT_MS = 15_000; // After a run closes, the mirror's commit can trail its last frame. const SETTLE_RELOAD_MS = [1_000, 3_000, 7_000, 15_000]; const NO_WAITS: readonly Wait[] = []; @@ -1016,14 +1021,19 @@ export function StageWorkspace({ setMobilePane("conversation"); setChangesAsked(true); }; - const composerDisabled = sending || approvalBusy || providerMissing; + const unbound = snapshot !== undefined && events === null; + const composerDisabled = + sending || approvalBusy || providerMissing || unbound; useEffect(() => { if (!changesAsked || composerDisabled) return; setChangesAsked(false); composerRef.current?.focus(); }, [changesAsked, composerDisabled]); - const unbound = snapshot !== undefined && events === null; + // A project that has talked before is between coordinators (an upgrade, a reopen): its thread + // stays and reads as the coordinator at work, not as a project starting from nothing. + const between = + unbound && mirror.length > 0 && startFault === null && modelNeeded === null; const startCoordinator = async () => { setStartFault(null); setModelNeeded(null); @@ -1045,6 +1055,20 @@ export function StageWorkspace({ else if (status === "archived" && room.data !== undefined) void room.refetch(); }, [projectId, status]); + const awaitingBind = + status === "active" && + unbound && + startFault === null && + modelNeeded === null; + useEffect(() => { + if (!awaitingBind) return; + const timer = setInterval(() => reloadRef.current(), BIND_POLL_MS); + const fallback = setTimeout(() => void startCoordinator(), BIND_WAIT_MS); + return () => { + clearInterval(timer); + clearTimeout(fallback); + }; + }, [awaitingBind]); const catalog = useModelCatalog(); const providerConnected = catalog.data !== undefined && !needsModelSetup(catalog.data.providers); @@ -1084,7 +1108,8 @@ export function StageWorkspace({ modelNeeded === null; // The mark stays for the whole turn: between model calls and during long tool work the live // reply shows nothing moving. - const thinking = !sendFailed && (sending || awaiting || (working && !held)); + const thinking = + !sendFailed && (sending || awaiting || (working && !held) || between); // Each concept's option is its own tab beside the others, never an ordinary document. const options = artifacts.filter((item) => item.kind === DESIGN_OPTION); const paneDeliverables = documentsForPane( @@ -1330,7 +1355,7 @@ export function StageWorkspace({ ); - } else if (unbound && !providerMissing) { + } else if (unbound && !providerMissing && !between) { // The person does not start the coordinator by typing: an unbound room waits for the // start that opening it asked for, and a refused start says why and offers it again. if (startFault === null) { diff --git a/apps/web/src/pages/workspace/load-room.ts b/apps/web/src/pages/workspace/load-room.ts index 3ee950c2..9dbf90cc 100644 --- a/apps/web/src/pages/workspace/load-room.ts +++ b/apps/web/src/pages/workspace/load-room.ts @@ -242,7 +242,10 @@ export async function loadProjectRoom( ]); fresh = { conversation: { - events: conversation?.events ?? null, + events: + conversation === null || conversation.runId === null + ? null + : conversation.events, turns: conversation?.turns ?? [], threads: conversation?.threads ?? [], runId: conversation?.runId ?? null, diff --git a/packages/schemas/src/conversation.ts b/packages/schemas/src/conversation.ts index 0ca75287..3ed80636 100644 --- a/packages/schemas/src/conversation.ts +++ b/packages/schemas/src/conversation.ts @@ -69,7 +69,7 @@ const conversationThread = type({ turns: turnList, }); export const ConversationEvents = type({ - runId: "string", + runId: "string | null", events: type({ seq: "number", type: "string",