diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 8022c89a..5ec24836 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -12,11 +12,11 @@ concurrency: permissions: contents: read -# Replace this marker with the accepted companion commit immediately after the -# companion PR creator commits it. Source-built jobs deliberately fail until it -# is one exact 40-character SHA; mutable refs and repository variables are forbidden. +# Keep the baseline daemon current independently of optional daemon capabilities. +# Both baseline and affinity-capable sources are immutable, separately tested commits. env: PASEO_E2E_COMMIT: 433e67b18b7964a92d593bdc78c518143accfc8b + PASEO_AFFINITY_E2E_COMMIT: bb3a46346c7e781bc963ffb2e0fdbd3f5ed3efa3 jobs: typecheck: @@ -168,6 +168,14 @@ jobs: run: npx playwright install --with-deps chromium - name: Run built-app browser tests run: npm run test:e2e:browser + - name: Attach workspace-affinity editor evidence + if: always() + uses: actions/upload-artifact@ea165f8d65b6e75b540449e92b4886f43607fa02 # v4 + with: + name: workspace-affinity-editor + path: e2e/screenshots/triggers/affinity-*.png + if-no-files-found: ignore + retention-days: 14 docker-smoke: runs-on: ubuntu-latest @@ -176,6 +184,41 @@ jobs: - name: Build and smoke-test production image run: npm run docker:smoke + workspace-affinity-e2e: + runs-on: ubuntu-latest + strategy: + fail-fast: false + matrix: + affinity: [false, true] + env: + PASEO_E2E_WORKTREE: ${{ github.workspace }}/paseo-source + PASEO_E2E_AFFINITY_SUPPORTED: ${{ matrix.affinity }} + steps: + - uses: actions/checkout@11d5960a326750d5838078e36cf38b85af677262 # v4 + - name: Require immutable baseline and affinity daemon commits + run: | + [[ "$PASEO_E2E_COMMIT" =~ ^[0-9a-f]{40}$ ]] + [[ "$PASEO_AFFINITY_E2E_COMMIT" =~ ^[0-9a-f]{40}$ ]] + - name: Check out exact Paseo source + uses: actions/checkout@11d5960a326750d5838078e36cf38b85af677262 # v4 + with: + repository: getpaseo/paseo + ref: ${{ matrix.affinity && env.PASEO_AFFINITY_E2E_COMMIT || env.PASEO_E2E_COMMIT }} + path: paseo-source + - uses: actions/setup-node@49933ea5288caeca8642d1e84afbd3f7d6820020 # v4 + with: + node-version: "22" + cache: "npm" + - name: Build exact Paseo source + working-directory: paseo-source + run: | + npm ci + npm run build:server + - run: npm ci + - run: npm run build + - name: Verify workspace affinity and legacy fallback + run: npm run test:e2e:hub:affinity + hub-e2e: runs-on: ubuntu-latest env: diff --git a/MAINTAINERS.md b/MAINTAINERS.md index 7c0df1f8..761f018d 100644 --- a/MAINTAINERS.md +++ b/MAINTAINERS.md @@ -18,12 +18,20 @@ npm run build npm run docker:smoke npm run test:e2e:browser npm run test:e2e:hub:source +npm run test:e2e:hub:affinity ``` -The source-built browser and Hub suites use the exact Paseo commit in `PASEO_E2E_COMMIT`. -When a Hub change depends on a Paseo protocol or CLI change, update that immutable SHA and -prove the combined contract before merging. Do not replace it with a branch or another mutable -reference. +The source-built browser and Hub suites use the exact baseline Paseo commit in `PASEO_E2E_COMMIT`. +When a Hub change depends on a Paseo protocol or CLI change, update that immutable SHA and prove +the combined contract before merging. Do not replace it with a branch or another mutable reference. + +Workspace affinity is optional, so CI also pins `PASEO_AFFINITY_E2E_COMMIT` independently. The +affinity suite runs against both commits: the baseline must complete executions with the existing +fresh-workspace behavior, while the capable daemon must retain, reuse, and restore a workspace +without reusing the agent. Set `PASEO_E2E_WORKTREE` to the built source checkout and +`PASEO_E2E_AFFINITY_SUPPORTED` to `false` or `true` respectively when running that suite locally. +Advance the baseline for unrelated daemon improvements without dropping the capable contract +test or forcing the unmerged affinity companion to carry those changes. Use the repository formatter through `npm run format` or `npm run format:files`. This repository uses Oxfmt, not Prettier. diff --git a/docs/public-api.md b/docs/public-api.md index 2f9ec3ac..e19dfd55 100644 --- a/docs/public-api.md +++ b/docs/public-api.md @@ -33,6 +33,58 @@ Hub sends each daemon the authored rendered prompt unchanged. Execution tools ar Daemon environments may author `worktree.newBranch: "trigger-${{ paseo.execution.id }}"` for a stable branch name unique to each agent execution. Hub materializes the execution UUID before persisting or dispatching the launch intent; recovery reuses that fully rendered intent. This is independent of whether a manual, Slack, Discord, GitHub, or Linear trigger selected the reusable environment. No prompt, context, input, value, step output, or provider event namespace is available in environment configuration, and unsupported expressions fail bundle activation at the authored `newBranch` field. +## Workspace affinity + +Workspace affinity is an explicit, step-scoped opt-in for a daemon workspace shared by related +trigger arrivals. For example, an agent that replies to Slack threads can retain one workspace per +thread: + +```yaml +triggers: + - name: respond-to-slack + on: slack.mention + filters: + from_users: ["U0123456789"] + max_runtime: 2h + steps: + - id: respond + environment: runner + max_runtime: 30m + idle_timeout: 5m + auto_archive: true + workspace_affinity: + key: "slack-thread:${{ paseo.trigger.conversation_key }}" + agent: { provider: codex } + prompt: [{ text: "Respond to the thread." }] +``` + +`paseo.trigger.conversation_key` is a provider-authenticated identifier, not prompt text. Slack +keys identify the connection, workspace, channel, and root thread; Discord keys identify the +connection, guild, channel, and thread (or its starter message); GitHub keys identify the +connection, repository, issue or pull request, and number. A key may also be a literal custom key, +or be composed from finite declared inputs, values, and outputs. Prompt text, ambient context, the +execution ID, and unbounded values are rejected so an untrusted event cannot select an existing +workspace. + +Hub sends only the opaque key and the triggering workflow's `max_runtime` deadline. Daemons that +support workspace affinity acknowledge the request, hash and persist their key-to-workspace +mapping, and reuse an active workspace or restore it if it was archived. Each matching arrival +extends retention to the later workflow deadline. With `auto_archive: true`, a supporting daemon +archives the workspace at that retained deadline—not at `idle_timeout`. `idle_timeout` remains an +in-execution liveness deadline. With `auto_archive: false`, Hub does not request affinity-driven +workspace archiving. + +Workspace affinity is a progressive daemon capability. Older daemons safely ignore the optional +request and continue creating and archiving fresh workspaces with their existing behavior; Hub does +not reject those executions or require an immediate daemon upgrade. Exact reuse, retention, and +archived-workspace restoration begin after the daemon is updated to a version that acknowledges +workspace affinity. + +Affinity does not serialize matching executions: each gets its own agent in the shared workspace. +All uses of a key must keep the same daemon target, cwd, worktree target, and auto-archive policy; +the daemon rejects a mismatch rather than mixing state. In particular, an execution-ID-derived +worktree branch is incompatible with reuse and is rejected during configuration compilation. + `deliveryKey` is caller-supplied request identity for the existing durable manual-event path. Hub namespaces it by the authenticated organization and resolved project before persistence, so the same caller key can be used independently in different tenants or projects. Existing receipt/run de-duplication applies, but this API does not promise exactly-once execution or guaranteed response replay; retries can still fail or conflict during restart and timing races. A successful representation contains `deliveryKey`, `providerEventReceiptId`, `triggerRunId`, `configuredTriggerName`, and the durable `workflowStatus`. The self-hosted Scalar reference is served with a restrictive Content Security Policy and does not require external fonts, scripts, telemetry, registries, or proxies. diff --git a/docs/trigger-migration.md b/docs/trigger-migration.md index 887f487f..483740d2 100644 --- a/docs/trigger-migration.md +++ b/docs/trigger-migration.md @@ -7,6 +7,7 @@ migration finishes before provider events are accepted and is safe to retry afte - Event type, filters, connection routing, and invocation inputs - Daemon, working directory, and worktree behavior +- Workspace affinity keys, retention deadlines, and auto-archive policy - Agent provider configuration and finite agent selection - The rendered prompt text, including resolved prompt partial content - Environment variables, GitHub authority, structured output, output grants, and timeouts @@ -18,6 +19,7 @@ migration finishes before provider events are accepted and is safe to retry afte These affect authoring or presentation, not what the active trigger is allowed to do: - YAML comments, whitespace, key ordering, anchors, and quoting style are regenerated. + String values such as an affinity key retain their exact bytes, including surrounding whitespace. - Prompt partial boundaries and file names disappear after their resolved content is inlined. - Shared environment and agent names disappear after their values are inlined. - A converted one-run workflow uses the internal step ID `run`; the former step ID remains only in diff --git a/docs/workflow-authority.md b/docs/workflow-authority.md index effe5499..41b6b7e8 100644 --- a/docs/workflow-authority.md +++ b/docs/workflow-authority.md @@ -3,6 +3,46 @@ Workflow authority is authored on an individual step. It is not a trigger option, agent option, sandbox setting, or Paseo daemon feature. +## Workspace selection + +An affinity key can select files left by earlier work, so it is an authority-bearing field. +Both `run.workspace_affinity.key` in a trigger document and `steps[].workspace_affinity.key` in a +legacy bundle use the same compiler validation. The editor preserves the authored key exactly and +never enables affinity by default. Prompt/context text cannot choose an existing workspace. + +Keep four identities separate: + +| Identity | Owns | Continuity | +| ---------------------- | ------------------------------------------------ | ----------------------------------------- | +| Workspace affinity key | Checkout and files | Related trigger arrivals | +| Hub execution ID | Completion, output grants, deadline, and retries | One step execution | +| External session ID | Provider reply destination and lifecycle signals | For example, one Linear Agent Session | +| Provider agent ID | Provider conversation/history | A new agent per execution in this feature | + +Linear's issue UUID is the conversation boundary, scoped by the authenticated connection and +Linear organization. Mutable identifiers, comment IDs, and delivery IDs cannot fragment that +identity. The Agent Sessions integration in [#88](https://github.com/getpaseo/hub/pull/88) can use +the same identity fields, but must add its implemented event names to conversation-key validation +and test multiple sessions on one issue, different issues/connections, and session-scoped replies +and cancellation. This PR does not enable unimplemented session events or resume provider agents. + +Hub owns configuration validation, trusted provider identity, and the durable launch intent. +The daemon owns atomic workspace binding, restoration, and safe expiry. Hub's restricted +`hub.execute` permission cannot restore existing workspaces on an older daemon. Requesting general +workspace-management authority to emulate affinity would broaden the operator's permissions and +duplicate daemon-local lifecycle state; do not do that implicitly. + +The latest arriving workflow deadline retains the workspace across gaps between executions. +Execution `idle_timeout` cannot represent those gaps because no agent need be running then. +Affinity does not serialize writers or preserve uncommitted files through destructive archival. +Those guarantees need separate, explicit policies; a shared key must not silently enable either. + +An old daemon may successfully execute a launch while ignoring affinity. Keep the baseline and +capable-daemon integration tests separate, as described in [MAINTAINERS.md](../MAINTAINERS.md). +The optional `workspaceAffinityApplied` acknowledgement is accepted on the wire but is not yet +retained in execution state. Operator-visible application status remains follow-up work: missing +acknowledgement must be distinguished from a confirmed application, including after reconnect. + ## Generic connection values Step environment values may explicitly request a named value from a configured diff --git a/e2e/helpers/hub.ts b/e2e/helpers/hub.ts index cb1877d3..61ecefc7 100644 --- a/e2e/helpers/hub.ts +++ b/e2e/helpers/hub.ts @@ -1637,11 +1637,16 @@ export class PaseoHub { this.hubCredentials.set(alias, credential); } - async approveDaemon(alias: string, displayName: string): Promise { + async approveDaemon( + alias: string, + displayName: string, + permissions: readonly string[] = [], + ): Promise { const credential = this.requireHubCredential(alias); const result = await this.requireSourcePaseo().connectWithCredential( this.primary.origin, credential, + permissions, ); const daemonId = z.string().uuid().parse(result["daemonId"]); await this.queryDatabase(this.primary, "update daemons set slug = $2 where id = $1", [ diff --git a/e2e/helpers/triggers.ts b/e2e/helpers/triggers.ts index 1aef0648..2718fc6f 100644 --- a/e2e/helpers/triggers.ts +++ b/e2e/helpers/triggers.ts @@ -445,6 +445,21 @@ export class OrganizationTriggers { await this.page.getByLabel("Instructions", { exact: true }).fill(prompt); } + async changeWorkspaceAffinity(key: string) { + const toggle = this.page.getByRole("button", { name: "Workspace affinity", exact: false }); + if ((await toggle.getAttribute("aria-expanded")) !== "true") await toggle.click(); + await this.page.getByLabel("Affinity key", { exact: true }).fill(key); + await expect( + this.page + .getByRole("alert") + .filter({ hasText: "Workspace sharing, not session continuation" }), + ).toContainText("Older daemons ignore affinity and create fresh workspaces."); + } + + async expectWorkspaceAffinity(key: string) { + await expect(this.page.getByLabel("Affinity key", { exact: true })).toHaveValue(key); + } + async expectMergeTagsAndAutosizing() { const instructions = this.page.getByLabel("Instructions", { exact: true }); await instructions.fill("Start finish"); diff --git a/e2e/triggers.spec.ts b/e2e/triggers.spec.ts index 490c0b86..bc0f6d6d 100644 --- a/e2e/triggers.spec.ts +++ b/e2e/triggers.spec.ts @@ -132,6 +132,55 @@ test("creates a trigger visually, preserves advanced YAML through the form, and }); }); +test("authors, persists, and explicitly disables workspace affinity in the trigger form", async ({ + hub, + page, +}) => { + await hub.signUpAs("owner", owner); + await hub.createOrganization("owner", "Acme"); + const daemon = await hub.connectProviderDaemon("owner", "Acme"); + await hub.seedSlackConnection("owner", "company-slack", "Acme Slack"); + const triggers = new OrganizationTriggers(page); + const key = " triage:${{ paseo.trigger.conversation_key }} "; + + await triggers.open(); + await triggers.startNew(); + await triggers.configureSlackMention({ + name: "affinity", + connection: "company-slack", + daemon, + cwd: "/workspace/acme", + users: "U123", + agent: "pi/gateway/vendor/model-v1", + mode: "full-access", + thinking: "high", + providerOptions: "", + prompt: "Handle the request.", + }); + await triggers.changeWorkspaceAffinity(key); + await page.getByRole("radio", { name: "New agent", exact: true }).click(); + await triggers.capture(`${SHOTS}/affinity-desktop.png`); + await page.setViewportSize({ width: 390, height: 844 }); + await page.getByLabel("Affinity key", { exact: true }).scrollIntoViewIfNeeded(); + await triggers.capture(`${SHOTS}/affinity-mobile.png`); + await page.setViewportSize({ width: 1280, height: 720 }); + await triggers.save("affinity"); + await triggers.openTrigger("affinity"); + await triggers.expectWorkspaceAffinity(key); + await triggers.changePrompt("Updated instructions must not reset the workspace key."); + await triggers.save("affinity"); + await triggers.openTrigger("affinity"); + await triggers.expectWorkspaceAffinity(key); + await triggers.switchToYaml(); + await triggers.expectYamlContains("workspace_affinity:", key); + await page.getByRole("radio", { name: "Form", exact: true }).click(); + await triggers.changeWorkspaceAffinity(""); + await triggers.save("affinity"); + await triggers.openTrigger("affinity"); + await triggers.switchToYaml(); + await expect(page.locator(".cm-content")).not.toContainText("workspace_affinity"); +}); + for (const scenario of [ { event: "github.issue_label_added", name: "issue-label" }, { event: "github.pull_request_label_added", name: "pr-label" }, diff --git a/package.json b/package.json index 7e029975..83153dc0 100644 --- a/package.json +++ b/package.json @@ -37,6 +37,7 @@ "test:scripts": "node --test scripts/release-metadata.node-test.mjs scripts/paseo-workspace.node-test.mjs", "test:release": "node --test scripts/release-metadata.node-test.mjs", "test:e2e:hub:source": "RUN_HUB_E2E=1 vitest run src/e2e/hub-foundation.e2e.test.ts src/e2e/hub-attachments.e2e.test.ts --bail=1", + "test:e2e:hub:affinity": "RUN_HUB_AFFINITY_E2E=1 vitest run src/e2e/hub-workspace-affinity.e2e.test.ts --bail=1", "test:e2e:hub:real-agent": "RUN_HUB_REAL_AGENT_E2E=1 vitest run src/e2e/hub-real-agent.e2e.test.ts --bail=1", "test:e2e:browser": "npm run build && playwright test", "format": "oxfmt .", diff --git a/src/config/compiler.test.ts b/src/config/compiler.test.ts index 1b95643e..2404aa42 100644 --- a/src/config/compiler.test.ts +++ b/src/config/compiler.test.ts @@ -120,6 +120,79 @@ describe("workflow compiler", () => { ); }); + it("rejects execution-scoped worktree branches for workspace affinity steps", () => { + const trigger = configuration().triggers[0]!; + const step = trigger.steps[0]!; + const executionScopedEnvironment = { + ...environment, + worktree: { + mode: "branch-off" as const, + newBranch: "trigger-${{ paseo.execution.id }}", + }, + }; + assert.throws( + () => + compileHubConfig({ + ...configuration(), + environments: [executionScopedEnvironment], + triggers: [ + { + ...trigger, + steps: [ + { + ...step, + workspace_affinity: { key: "shared-review" }, + }, + ], + }, + ], + }), + (error) => { + assert.ok(error instanceof Error); + assert.deepEqual(Reflect.get(error, "path"), [ + "triggers", + "run", + "steps", + "work", + "workspace_affinity", + ]); + assert.match(error.message, /stable worktree target.*paseo\.execution\.id/iu); + return true; + }, + ); + + assert.throws( + () => + compileHubConfig({ + ...configuration(), + environments: [ + environment, + { ...executionScopedEnvironment, name: "execution-scoped-runner" }, + ], + triggers: [ + { + ...trigger, + inputs: { + runner: { + type: "string", + required: true, + choices: ["runner", "execution-scoped-runner"], + }, + }, + steps: [ + { + ...step, + environment: "${{ paseo.inputs.runner }}", + workspace_affinity: { key: "shared-review" }, + }, + ], + }, + ], + }), + /environment execution-scoped-runner.*stable worktree target/iu, + ); + }); + it("preserves opaque provider options and leaves an omitted mode omitted", () => { const sourceOptions = { sandbox_workspace_write: { @@ -621,6 +694,293 @@ describe("workflow compiler", () => { ); }); + it("allows explicit workspace affinity keys without letting prompt text select a workspace", () => { + const trigger = configuration().triggers[0]!; + const step = trigger.steps[0]!; + const conversationTrigger = { + ...trigger, + on: "slack.mention", + filters: { from_users: ["U_ALLOWED"] }, + }; + const compiled = compileHubConfig({ + ...configuration(), + triggers: [ + { + ...conversationTrigger, + steps: [ + { + ...step, + workspace_affinity: { + key: " review-${{ paseo.trigger.conversation_key }} ", + }, + }, + ], + }, + ], + }); + assert.deepEqual(compiled.triggers[0]?.steps[0]?.workspaceAffinity, { + key: " review-${{ paseo.trigger.conversation_key }} ", + }); + assert.deepEqual(parseCompiledHubConfig(compiled), compiled); + + const renderedBoundedKey = "x".repeat(505) + "${{ true }}"; + assert.ok(renderedBoundedKey.length > 512); + const compiledLongTemplate = compileHubConfig({ + ...configuration(), + triggers: [ + { + ...conversationTrigger, + steps: [{ ...step, workspace_affinity: { key: renderedBoundedKey } }], + }, + ], + }); + assert.deepEqual(compiledLongTemplate.triggers[0]?.steps[0]?.workspaceAffinity, { + key: renderedBoundedKey, + }); + assert.deepEqual(parseCompiledHubConfig(compiledLongTemplate), compiledLongTemplate); + + assert.throws( + () => + compileHubConfig({ + ...configuration(), + triggers: [ + { + ...trigger, + steps: [ + { + ...step, + workspace_affinity: { + key: "review-${{ paseo.trigger.conversation_key }}", + }, + }, + ], + }, + ], + }), + /manual\.run does not provide a conversation key/iu, + ); + + assert.throws( + () => + compileHubConfig({ + ...configuration(), + triggers: [ + { + ...conversationTrigger, + on: "github.push", + steps: [ + { + ...step, + workspace_affinity: { + key: "review-${{ paseo.trigger.conversation_key }}", + }, + }, + ], + }, + ], + }), + /github\.push does not provide a conversation key/iu, + ); + + assert.doesNotThrow(() => + compileHubConfig({ + ...configuration(), + triggers: [ + { + ...trigger, + steps: [{ ...step, workspace_affinity: { key: "shared-release-triage" } }], + }, + ], + }), + ); + assert.throws( + () => + compileHubConfig({ + ...configuration(), + triggers: [ + { + ...trigger, + steps: [{ ...step, workspace_affinity: { key: " " } }], + }, + ], + }), + /workspace affinity key must not be blank/iu, + ); + assert.throws( + () => + compileHubConfig({ + ...configuration(), + triggers: [ + { + ...trigger, + steps: [{ ...step, workspace_affinity: { key: "${{ paseo.prompt }}" } }], + }, + ], + }), + /paseo\.prompt.*authority-bearing/iu, + ); + }); + + it("propagates workspace affinity authority checks through referenced values", () => { + const trigger = configuration().triggers[0]!; + const step = trigger.steps[0]!; + + assert.throws( + () => + compileHubConfig({ + ...configuration(), + triggers: [ + { + ...trigger, + values: { + prompt_route: "${{ paseo.prompt == 'reuse' }}", + route: "${{ values.prompt_route }}", + }, + steps: [{ ...step, workspace_affinity: { key: "shared-${{ values.route }}" } }], + }, + ], + }), + /paseo\.prompt.*authority-bearing/iu, + ); + + assert.throws( + () => + compileHubConfig({ + ...configuration(), + triggers: [ + { + ...trigger, + inputs: { route: { type: "string", required: true } }, + values: { route: "${{ paseo.inputs.route == 'reuse' }}" }, + steps: [{ ...step, workspace_affinity: { key: "shared-${{ values.route }}" } }], + }, + ], + }), + /input route without finite choices/iu, + ); + + assert.throws( + () => + compileHubConfig({ + ...configuration(), + triggers: [ + { + ...trigger, + values: { route: "${{ steps.classify.outputs.route == 'reuse' }}" }, + steps: [ + { + ...step, + id: "classify", + output: { + schema: { + type: "object", + properties: { route: { type: "string" } }, + }, + }, + }, + { + ...step, + id: "work", + workspace_affinity: { key: "shared-${{ values.route }}" }, + }, + ], + }, + ], + }), + /agent output without provable finite choices/iu, + ); + + assert.doesNotThrow(() => + compileHubConfig({ + ...configuration(), + triggers: [ + { + ...trigger, + inputs: { + route: { type: "string", required: true, choices: ["reuse", "fresh"] }, + }, + values: { route: "${{ paseo.inputs.route == 'reuse' }}" }, + steps: [{ ...step, workspace_affinity: { key: "shared-${{ values.route }}" } }], + }, + ], + }), + ); + + assert.doesNotThrow(() => + compileHubConfig({ + ...configuration(), + triggers: [ + { + ...trigger, + on: "slack.mention", + filters: { from_users: ["U_ALLOWED"] }, + values: { + provider_conversation: "${{ paseo.trigger.conversation_key }}", + conversation: "${{ values.provider_conversation }}", + }, + steps: [ + { + ...step, + workspace_affinity: { key: "shared-${{ values.conversation }}" }, + }, + ], + }, + ], + }), + ); + + assert.throws( + () => + compileHubConfig({ + ...configuration(), + triggers: [ + { + ...trigger, + on: "slack.mention", + filters: { from_users: ["U_ALLOWED"] }, + values: { + provider_conversation: "${{ paseo.trigger.conversation_key }}", + conversation: "${{ values.provider_conversation }}", + }, + steps: [ + { + ...step, + prompt: [{ text: "Conversation: ${{ values.conversation }}" }], + workspace_affinity: { key: "shared-${{ values.conversation }}" }, + }, + ], + }, + ], + }), + /conversation_key outside workspace_affinity\.key/iu, + ); + }); + + it("rejects conversation-key value graphs without workspace-affinity consumers", () => { + const trigger = configuration().triggers[0]!; + const step = trigger.steps[0]!; + + assert.throws( + () => + compileHubConfig({ + ...configuration(), + triggers: [ + { + ...trigger, + on: "slack.mention", + filters: { from_users: ["U_ALLOWED"] }, + values: { + provider_conversation: "${{ paseo.trigger.conversation_key }}", + conversation: "${{ values.provider_conversation }}", + }, + steps: [step], + }, + ], + }), + /conversation_key outside workspace_affinity\.key/iu, + ); + }); + it("rejects the removed prompt inventory compatibility key", () => { const trigger = configuration().triggers[0]!; assert.throws( diff --git a/src/config/compiler.ts b/src/config/compiler.ts index bc83b711..060ffa9d 100644 --- a/src/config/compiler.ts +++ b/src/config/compiler.ts @@ -4,6 +4,7 @@ import { createHash } from "node:crypto"; import { z } from "zod"; import { expressionPaths, + expressionPathsInTemplate, parseExpression, validateExecutionTemplate, type Expression, @@ -25,6 +26,8 @@ import { type CompiledGitHubAuthority, } from "./github-authority.js"; import { validateConnectionTemplate } from "./connection-template.js"; +import { WorkspaceAffinityKeySchema, WorkspaceAffinitySchema } from "./workspace-affinity.js"; +import { triggerSupportsWorkspaceAffinityConversationKey } from "../triggers/workspace-affinity.js"; const IDENTIFIER = /^[a-z][a-z0-9_-]*$/u; const EVENT_NAME = /^[a-z][a-z0-9_-]*\.[a-z][a-z0-9_-]*$/u; @@ -177,6 +180,7 @@ const StepSchema = z output: z.object({ schema: JsonSchemaSchema }).strict().optional(), allow_outputs: z.array(AllowOutputSchema).optional(), auto_archive: z.boolean().optional(), + workspace_affinity: WorkspaceAffinitySchema.optional(), }) .strict(); @@ -238,6 +242,10 @@ export interface CompiledInput { choices?: readonly JsonPrimitive[] | undefined; } +export interface CompiledWorkspaceAffinity { + key: string; +} + export interface CompiledStep { continuation?: import("../triggers/continuation.js").Continuation | undefined; id: string; @@ -253,6 +261,7 @@ export interface CompiledStep { output?: { schema: JsonValue } | undefined; allowOutputs: readonly { type: string; max?: number | undefined; required: boolean }[]; autoArchive: boolean; + workspaceAffinity?: CompiledWorkspaceAffinity | undefined; } export type CompiledSteps = readonly CompiledStep[]; @@ -353,6 +362,10 @@ const CompiledInputSchema: z.ZodType = z }) .strict(); +const CompiledWorkspaceAffinitySchema: z.ZodType = z + .object({ key: WorkspaceAffinityKeySchema }) + .strict(); + const CompiledEnvironmentSchema = z.discriminatedUnion("kind", [ z .object({ @@ -410,6 +423,7 @@ const CompiledStepSchema: z.ZodType = z .strict(), ), autoArchive: z.boolean(), + workspaceAffinity: CompiledWorkspaceAffinitySchema.optional(), }) .strict(); @@ -616,6 +630,9 @@ function compileStep( required: allowOutput.required ?? false, })), autoArchive: step.auto_archive ?? false, + ...(step.workspace_affinity === undefined + ? {} + : { workspaceAffinity: { key: step.workspace_affinity.key } }), }; } @@ -844,8 +861,19 @@ function validateExpressionContract( const stepOrdinals = new Map(trigger.steps.map((step, ordinal) => [step.id, ordinal])); const valueNames = new Set(Object.keys(trigger.values)); const visiting = new Set(); + const workspaceAffinityValues = new Set(); - for (const name of valueNames) validateValue(name); + for (const step of trigger.steps) { + if (step.workspaceAffinity === undefined) continue; + compileAt(["triggers", triggerName, "steps", step.id, "workspace_affinity", "key"], () => { + for (const reference of expressionPathsInTemplate(step.workspaceAffinity!.key)) { + if (reference.namespace === "values") collectWorkspaceAffinityValue(reference.name); + } + }); + } + for (const name of valueNames) { + if (!workspaceAffinityValues.has(name)) validateValue(name); + } for (const [ordinal, step] of trigger.steps.entries()) { if (step.condition !== undefined) compileAt(["triggers", triggerName, "steps", step.id, "if"], () => @@ -870,6 +898,18 @@ function validateExpressionContract( } }); } + if (step.workspaceAffinity !== undefined) { + compileAt(["triggers", triggerName, "steps", step.id, "workspace_affinity", "key"], () => + validateWorkspaceAffinityTemplate( + step.workspaceAffinity!.key, + ordinal, + `step ${step.id} workspace_affinity.key`, + ), + ); + compileAt(["triggers", triggerName, "steps", step.id, "workspace_affinity"], () => + validateWorkspaceAffinityEnvironmentSelection(step.environment, ordinal, step.id), + ); + } for (const [index, block] of step.prompt.entries()) { compileAt(["triggers", triggerName, "steps", step.id, "prompt", index], () => validateTemplate( @@ -898,6 +938,41 @@ function validateExpressionContract( } } + function validateWorkspaceAffinityEnvironmentSelection( + template: string, + ordinal: number, + stepId: string, + ): void { + const selected = finiteTemplateValues(template, ordinal); + if (selected === undefined) return; + for (const name of selected) { + const environment = environments.get(name); + if (environment?.kind !== "daemon" || environment.worktree?.mode !== "branch-off") continue; + const executionScoped = expressionPathsInTemplate(environment.worktree.newBranch).some( + (reference) => + reference.namespace === "paseo" && + Array.isArray(reference.path) && + reference.path[0] === "execution" && + reference.path[1] === "id", + ); + if (executionScoped) { + throw new Error( + `step ${stepId} workspace affinity environment ${name} must use a stable worktree target; worktree.newBranch references paseo.execution.id`, + ); + } + } + } + + function collectWorkspaceAffinityValue(name: string): void { + if (workspaceAffinityValues.has(name)) return; + workspaceAffinityValues.add(name); + const expression = trigger.values[name]; + if (expression === undefined) return; + for (const reference of expressionPaths(expression)) { + if (reference.namespace === "values") collectWorkspaceAffinityValue(reference.name); + } + } + function finiteTemplateValues(template: string, ordinal: number): readonly string[] | undefined { let results = [""]; let cursor = 0; @@ -964,12 +1039,20 @@ function validateExpressionContract( return output === undefined ? undefined : finiteSchemaChoices(output.schema, reference.path); } - function validateValue(name: string, ordinal = Number.POSITIVE_INFINITY): void { + function validateValue( + name: string, + ordinal = Number.POSITIVE_INFINITY, + mode: "ordinary" | "authority" | "affinity" = "ordinary", + ): void { if (visiting.has(name)) throw new Error(`value dependency cycle includes ${name}`); const expression = trigger.values[name]; if (expression === undefined) throw new Error(`value ${name} is unavailable`); visiting.add(name); - validateExpression(expression, ordinal, `value ${name}`, false); + if (mode === "affinity") { + validateWorkspaceAffinityExpression(expression, ordinal, `value ${name}`); + } else { + validateExpression(expression, ordinal, `value ${name}`, mode === "authority"); + } visiting.delete(name); } @@ -982,7 +1065,9 @@ function validateExpressionContract( ): void { for (const reference of expressionPaths(expression)) { validateReference(reference, ordinal, path, authorityBearing, contextAllowed); - if (reference.namespace === "values") validateValue(reference.name, ordinal); + if (reference.namespace === "values") { + validateValue(reference.name, ordinal, authorityBearing ? "authority" : "ordinary"); + } } if (authorityBearing && !isFiniteAuthorityExpression(expression)) { throw new Error( @@ -1011,32 +1096,51 @@ function validateExpressionContract( } } + function validateWorkspaceAffinityTemplate(value: string, ordinal: number, path: string): void { + let cursor = 0; + while (true) { + const start = value.indexOf(EXPRESSION_START, cursor); + if (start < 0) return; + const end = value.indexOf(EXPRESSION_END, start + EXPRESSION_START.length); + if (end < 0) throw new Error(`${path} uses an unterminated expression`); + const expression = parseExpression(value.slice(start + EXPRESSION_START.length, end)); + validateWorkspaceAffinityExpression(expression, ordinal, path); + cursor = end + EXPRESSION_END.length; + } + } + + function validateWorkspaceAffinityExpression( + expression: Expression, + ordinal: number, + path: string, + ): void { + for (const reference of expressionPaths(expression)) { + validateReference(reference, ordinal, path, true, false, true); + if (reference.namespace === "values") validateValue(reference.name, ordinal, "affinity"); + } + if (!isAffinityKeyExpression(expression)) { + throw new Error( + `${path} uses an unbounded key; use paseo.trigger.conversation_key or finite choices`, + ); + } + } + function validateReference( reference: ExpressionPath, ordinal: number, path: string, authorityBearing: boolean, contextAllowed: boolean, + conversationKeyAllowed = false, ): void { if (reference.namespace === "paseo") { - if (reference.path === "prompt") { - if (authorityBearing) - throw new Error(`${path} uses paseo.prompt in an authority-bearing field`); - return; - } - if (reference.path === "context") { - if (!contextAllowed) throw new Error(`${path} uses paseo.context outside a step prompt`); - return; - } - if (reference.path[0] === "execution") { - throw new Error(`${path} uses paseo.execution outside environment worktree.newBranch`); - } - const inputName = reference.path[1]; - const input = trigger.inputs[inputName]; - if (input === undefined) throw new Error(`${path} references undeclared input ${inputName}`); - if (authorityBearing && input.choices === undefined) { - throw new Error(`${path} uses input ${inputName} without finite choices`); - } + validatePaseoReference( + reference, + path, + authorityBearing, + contextAllowed, + conversationKeyAllowed, + ); return; } if (reference.namespace === "values") { @@ -1063,6 +1167,50 @@ function validateExpressionContract( } } + function validatePaseoReference( + reference: Extract, + path: string, + authorityBearing: boolean, + contextAllowed: boolean, + conversationKeyAllowed: boolean, + ): void { + if (reference.path === "prompt") { + if (authorityBearing) + throw new Error(`${path} uses paseo.prompt in an authority-bearing field`); + return; + } + if (reference.path === "context") { + if (!contextAllowed) throw new Error(`${path} uses paseo.context outside a step prompt`); + return; + } + if (reference.path[0] === "execution") { + throw new Error(`${path} uses paseo.execution outside environment worktree.newBranch`); + } + if (reference.path[0] === "trigger") { + if ( + !conversationKeyAllowed || + reference.path[1] !== "conversation_key" || + reference.path.length !== 2 + ) { + throw new Error( + `${path} uses paseo.trigger.conversation_key outside workspace_affinity.key`, + ); + } + if (!triggerSupportsWorkspaceAffinityConversationKey(trigger.on)) { + throw new Error( + `${path} uses paseo.trigger.conversation_key, but trigger event ${trigger.on} does not provide a conversation key`, + ); + } + return; + } + const inputName = reference.path[1]; + const input = trigger.inputs[inputName]; + if (input === undefined) throw new Error(`${path} references undeclared input ${inputName}`); + if (authorityBearing && input.choices === undefined) { + throw new Error(`${path} uses input ${inputName} without finite choices`); + } + } + function isFiniteAuthorityExpression(expression: Expression): boolean { if (expression.kind === "literal") return expression.value !== null; if (expression.kind === "not") return true; @@ -1088,6 +1236,28 @@ function validateExpressionContract( return step?.output !== undefined && hasFiniteSchemaChoices(step.output.schema, reference.path); } + function isAffinityKeyExpression(expression: Expression): boolean { + if (expression.kind === "literal") return true; + if (expression.kind === "not") return isAffinityKeyExpression(expression.value); + if (expression.kind === "binary") { + return isAffinityKeyExpression(expression.left) && isAffinityKeyExpression(expression.right); + } + const reference = expression.value; + if ( + reference.namespace === "paseo" && + Array.isArray(reference.path) && + reference.path[0] === "trigger" && + reference.path[1] === "conversation_key" + ) { + return true; + } + if (reference.namespace === "values") { + const value = trigger.values[reference.name]; + return value !== undefined && isAffinityKeyExpression(value); + } + return isFiniteAuthorityExpression(expression); + } + if ( trigger.steps.some((step) => step.environment.includes("${{")) && environmentNames.size === 0 @@ -1246,6 +1416,7 @@ function validateStepEnvironmentContract( } } } + if (github !== undefined) { validateGitHubAuthority(github, `trigger ${triggerName} step ${stepId} github`); if (github.repositories === undefined && !triggerEvent.startsWith("github.")) { @@ -1319,7 +1490,8 @@ function isExpressionPath(value: unknown): boolean { (Array.isArray(value["path"]) && value["path"].length === 2 && ((value["path"][0] === "inputs" && typeof value["path"][1] === "string") || - (value["path"][0] === "execution" && value["path"][1] === "id")))) + (value["path"][0] === "execution" && value["path"][1] === "id") || + (value["path"][0] === "trigger" && value["path"][1] === "conversation_key")))) ); } diff --git a/src/config/workspace-affinity.ts b/src/config/workspace-affinity.ts new file mode 100644 index 00000000..9bc45e4f --- /dev/null +++ b/src/config/workspace-affinity.ts @@ -0,0 +1,10 @@ +import { z } from "zod"; + +// This is an authored template, not the rendered key. Runtime enforces the 512-character bound. +// Never normalize it: leading/trailing whitespace can be part of an existing workspace identity. +export const WorkspaceAffinityKeySchema = z + .string() + .min(1) + .refine((key) => key.trim().length > 0, "workspace affinity key must not be blank"); + +export const WorkspaceAffinitySchema = z.object({ key: WorkspaceAffinityKeySchema }).strict(); diff --git a/src/daemons/agents/index.test.ts b/src/daemons/agents/index.test.ts index 3b594fd6..8dba085a 100644 --- a/src/daemons/agents/index.test.ts +++ b/src/daemons/agents/index.test.ts @@ -64,6 +64,43 @@ test("requires ordinary agent RPCs and durable receipts instead of falling back expect(frames).toEqual([]); }); +test("forwards optional workspace affinity through the ordinary agent creation request", async () => { + const requests: Record[] = []; + const agents = new DaemonAgents((frame) => { + const { message } = z + .object({ message: z.record(z.string(), z.unknown()) }) + .parse(JSON.parse(frame)); + requests.push(message); + agents.receive({ + type: "session", + message: { + type: "status", + payload: { + requestId: message["requestId"], + status: "agent_created", + agent: { id: "agent", workspaceId: "workspace", status: "idle" }, + }, + }, + }); + }); + enable(agents); + await agents.create("affinity-key", { + ...options, + workspaceAffinity: { + key: "thread-1", + retainUntil: "2026-08-06T12:02:00.000Z", + autoArchive: true, + }, + }); + expect(requests[0]).toMatchObject({ + workspaceAffinity: { + key: "thread-1", + retainUntil: "2026-08-06T12:02:00.000Z", + autoArchive: true, + }, + }); +}); + test("a lost response remains recoverable instead of reporting a rejected creation", async () => { const agents = new DaemonAgents(() => {}); enable(agents); diff --git a/src/daemons/agents/index.ts b/src/daemons/agents/index.ts index a2278364..d6253e99 100644 --- a/src/daemons/agents/index.ts +++ b/src/daemons/agents/index.ts @@ -130,6 +130,7 @@ export class DaemonAgents implements AgentConnection { }, env: options.env, worktree: options.worktree, + workspaceAffinity: options.workspaceAffinity, }, timeoutMs, ); diff --git a/src/daemons/lifecycle.ts b/src/daemons/lifecycle.ts index 0cf2e7d0..ddbfd7e4 100644 --- a/src/daemons/lifecycle.ts +++ b/src/daemons/lifecycle.ts @@ -1958,6 +1958,9 @@ async function buildCreateAgentOptions( : { worktree: intent.environment.worktree, }), + ...(intent.workspaceAffinity === undefined + ? {} + : { workspaceAffinity: intent.workspaceAffinity }), }; } diff --git a/src/daemons/protocol.ts b/src/daemons/protocol.ts index e4684b92..1db738bb 100644 --- a/src/daemons/protocol.ts +++ b/src/daemons/protocol.ts @@ -17,6 +17,11 @@ export interface DaemonCreateAgentOptions { env: Record; mcpServers?: Record; worktree?: WorktreeTarget; + workspaceAffinity?: { + key: string; + retainUntil: string; + autoArchive: boolean; + }; } export interface McpToolRef { diff --git a/src/daemons/test-utils/daemon-registry-harness.ts b/src/daemons/test-utils/daemon-registry-harness.ts index ddd77246..925d7695 100644 --- a/src/daemons/test-utils/daemon-registry-harness.ts +++ b/src/daemons/test-utils/daemon-registry-harness.ts @@ -51,7 +51,10 @@ export class DaemonRegistryHarness { return harness; } - async pendingCreate(executionId: string): Promise> { + async pendingCreate( + executionId: string, + options: { workspaceAffinity?: boolean } = {}, + ): Promise> { const connection = this.connection(); const promise = connection.agents.create(executionId, { provider: "opencode", @@ -62,6 +65,15 @@ export class DaemonRegistryHarness { toolPolicy: { preapproved: [{ kind: "mcp", server: "hub", tool: "finish_execution" }], }, + ...(options.workspaceAffinity + ? { + workspaceAffinity: { + key: "thread-1", + retainUntil: "2026-08-06T12:02:00.000Z", + autoArchive: true, + }, + } + : {}), }); void promise.catch(() => undefined); return { diff --git a/src/dispatcher/launch-machine-intent.test.ts b/src/dispatcher/launch-machine-intent.test.ts index d5f6e17f..07830426 100644 --- a/src/dispatcher/launch-machine-intent.test.ts +++ b/src/dispatcher/launch-machine-intent.test.ts @@ -29,6 +29,11 @@ describe("LaunchMachineIntent", () => { timeoutMs: 3_600_000, idleTimeoutMs: 300_000, autoArchive: true, + workspaceAffinity: { + key: "discord-thread-1", + retainUntil: "2026-08-06T12:02:00.000Z", + autoArchive: true, + }, triggerContext: { messageId: "message-1" }, outputContext: { messageId: "message-1" }, hubConfig: { triggers: [] }, @@ -59,6 +64,11 @@ describe("LaunchMachineIntent", () => { timeoutMs: 3_600_000, idleTimeoutMs: 300_000, autoArchive: true, + workspaceAffinity: { + key: "discord-thread-1", + retainUntil: "2026-08-06T12:02:00.000Z", + autoArchive: true, + }, triggerContext: { messageId: "message-1" }, outputContext: { messageId: "message-1" }, configurationRevisionId: "config-version-1", diff --git a/src/dispatcher/launch-machine-intent.ts b/src/dispatcher/launch-machine-intent.ts index 9fccd530..df3c3a48 100644 --- a/src/dispatcher/launch-machine-intent.ts +++ b/src/dispatcher/launch-machine-intent.ts @@ -15,6 +15,16 @@ export interface DaemonEnvironmentTarget { worktree?: WorktreeTarget; } +/** + * An opaque, daemon-scoped workspace lease. The daemon owns the resulting workspace mapping; + * Hub never receives or selects a workspace ID. + */ +export interface WorkspaceAffinity { + key: string; + retainUntil: string; + autoArchive: boolean; +} + export interface LaunchMachineIntent { continuation?: { key: string | null; compatibility: unknown }; kind: "launch_machine"; @@ -34,6 +44,7 @@ export interface LaunchMachineIntent { startupTimeoutMs?: number; idleTimeoutMs?: number; autoArchive: boolean; + workspaceAffinity?: WorkspaceAffinity; triggerContext: unknown; outputContext: unknown; outputSchema?: JsonValue; @@ -59,6 +70,7 @@ export function buildLaunchMachineIntent(input: { startupTimeoutMs?: number; idleTimeoutMs?: number; autoArchive: boolean; + workspaceAffinity?: WorkspaceAffinity; triggerContext: unknown; outputContext: unknown; hubConfig: unknown; @@ -80,6 +92,15 @@ export function buildLaunchMachineIntent(input: { ...(input.startupTimeoutMs === undefined ? {} : { startupTimeoutMs: input.startupTimeoutMs }), ...(input.idleTimeoutMs === undefined ? {} : { idleTimeoutMs: input.idleTimeoutMs }), autoArchive: input.autoArchive, + ...(input.workspaceAffinity === undefined + ? {} + : { + workspaceAffinity: { + key: input.workspaceAffinity.key, + retainUntil: input.workspaceAffinity.retainUntil, + autoArchive: input.workspaceAffinity.autoArchive, + }, + }), triggerContext: input.triggerContext, outputContext: input.outputContext, configurationRevisionId: input.configurationRevisionId, diff --git a/src/e2e/harness/acp-agent.mjs b/src/e2e/harness/acp-agent.mjs index e9b308e0..01b6fd9a 100644 --- a/src/e2e/harness/acp-agent.mjs +++ b/src/e2e/harness/acp-agent.mjs @@ -109,13 +109,18 @@ class PhaseFiveAgent { } if (service === "daemon-restart" || prompt.includes("daemon-restart")) await runUntilInterrupted(); - if (this.hubMcp && sessionExecutionId) { - const finish = await callMcp(this.hubMcp, 4, "tools/call", { - name: "finish_execution", - arguments: executionArgs, - }); - if (finish.result?.isError) throw new Error(JSON.stringify(finish)); - } else if (this.hubMcp) await scheduleHubCompletion(this.hubMcp); + if (this.hubMcp) { + if (sessionExecutionId !== undefined || process.env["HUB_E2E_COMPLETE_IN_TURN"] === "1") { + await initializeMcp(this.hubMcp); + const completion = await callMcp(this.hubMcp, 4, "tools/call", { + name: "finish_execution", + arguments: executionArgs, + }); + if (completion.result?.isError === true) throw new Error("Hub completion was rejected"); + } else { + await scheduleHubCompletion(this.hubMcp); + } + } return { stopReason: "end_turn" }; } } diff --git a/src/e2e/harness/fault-proxy.test.ts b/src/e2e/harness/fault-proxy.test.ts new file mode 100644 index 00000000..21beb57b --- /dev/null +++ b/src/e2e/harness/fault-proxy.test.ts @@ -0,0 +1,65 @@ +import assert from "node:assert/strict"; +import { once } from "node:events"; +import { createServer, type IncomingMessage, type Server } from "node:http"; +import { createServer as createNetServer } from "node:net"; +import { afterEach, describe, it } from "vitest"; +import { WebSocket, WebSocketServer } from "ws"; +import { HubFaultProxy } from "./fault-proxy.js"; + +describe("HubFaultProxy", () => { + let proxy: HubFaultProxy | undefined; + let targetServer: Server | undefined; + let targetSockets: WebSocketServer | undefined; + let daemonSocket: WebSocket | undefined; + + afterEach(async () => { + daemonSocket?.terminate(); + await proxy?.stop(); + for (const socket of targetSockets?.clients ?? []) socket.terminate(); + if (targetSockets) await new Promise((resolve) => targetSockets!.close(() => resolve())); + targetServer?.closeIdleConnections(); + targetServer?.closeAllConnections(); + if (targetServer) await new Promise((resolve) => targetServer!.close(() => resolve())); + }); + + it("preserves daemon session protocol negotiation across its terminated upgrade", async () => { + targetServer = createServer(); + targetSockets = new WebSocketServer({ server: targetServer }); + await new Promise((resolve, reject) => { + targetServer!.once("error", reject); + targetServer!.listen(0, "127.0.0.1", resolve); + }); + const targetAddress = targetServer.address(); + assert.ok(targetAddress && typeof targetAddress !== "string"); + + proxy = await HubFaultProxy.start( + `http://127.0.0.1:${targetAddress.port}`, + await availablePort(), + ); + daemonSocket = new WebSocket(`${proxy.origin.replace("http:", "ws:")}/api/daemons/socket`, { + headers: { "x-paseo-session-protocol": "1" }, + }); + const upgrade = new Promise((resolve) => + daemonSocket!.once("upgrade", resolve), + ); + + await once(daemonSocket, "open"); + + assert.equal((await upgrade).headers["x-paseo-session-protocol"], "1"); + daemonSocket.close(); + await once(daemonSocket, "close"); + daemonSocket = undefined; + }); +}); + +async function availablePort(): Promise { + const server = createNetServer(); + await new Promise((resolve, reject) => { + server.once("error", reject); + server.listen(0, "127.0.0.1", resolve); + }); + const address = server.address(); + assert.ok(address && typeof address !== "string"); + await new Promise((resolve) => server.close(() => resolve())); + return address.port; +} diff --git a/src/e2e/harness/fault-proxy.ts b/src/e2e/harness/fault-proxy.ts index 43a69f42..1f9c6a7b 100644 --- a/src/e2e/harness/fault-proxy.ts +++ b/src/e2e/harness/fault-proxy.ts @@ -3,6 +3,9 @@ import type { Duplex } from "node:stream"; import { WebSocket, WebSocketServer, type RawData } from "ws"; import { z } from "zod"; +const DAEMON_SESSION_PROTOCOL_HEADER = "x-paseo-session-protocol"; +const DAEMON_SESSION_PROTOCOL_VERSION = "1"; + interface Denial { type: "rpc_error"; requestType: string; @@ -39,8 +42,11 @@ export class HubFaultProxy { void this.forwardHttp(request, response); }); this.sockets.on("headers", (headers, request) => { - if (request.headers["x-paseo-session-protocol"] === "1") { - headers.push("x-paseo-session-protocol: 1"); + // The fault proxy terminates the daemon-side upgrade before it can observe the upstream + // response. Mirror Hub's opt-in negotiation acknowledgement so both sides agree whether + // the daemon session begins with a real hello or the legacy synthetic one. + if (request.headers[DAEMON_SESSION_PROTOCOL_HEADER] === DAEMON_SESSION_PROTOCOL_VERSION) { + headers.push(`${DAEMON_SESSION_PROTOCOL_HEADER}: ${DAEMON_SESSION_PROTOCOL_VERSION}`); } }); this.server.on("upgrade", (request, socket, head) => this.upgrade(request, socket, head)); diff --git a/src/e2e/harness/index.ts b/src/e2e/harness/index.ts index 21d1ce97..d5553db0 100644 --- a/src/e2e/harness/index.ts +++ b/src/e2e/harness/index.ts @@ -21,8 +21,11 @@ const HUB_ROOT = process.cwd(); let MACHINE_KEY = ""; const PROJECT_ID = "00000000-0000-4000-8000-000000000001"; const PROJECT_SLUG = "default"; +const SOURCE_E2E_DAEMON_ID = "00000000-0000-4000-8000-0000000000dd"; const PersistedDaemonAgentSchema = z.object({ id: z.string(), + workspaceId: z.string().optional(), + archivedAt: z.string().nullable().optional(), lastStatus: z.enum(["error", "initializing", "idle", "running", "closed"]), config: z.object({ mcpServers: z @@ -109,6 +112,7 @@ interface ManagedChild { interface HubE2EOptions { realAgent?: boolean; + completeInTurn?: boolean; } export interface SourceCliBundleDeploymentEvidence { @@ -181,6 +185,7 @@ export class HubE2E { status = await this.requireSource().connectWithCredential( this.requireProxy().origin, MACHINE_KEY, + ["hub.execute"], ); } catch (error) { const enrollmentState = await this.requirePool().query<{ @@ -208,6 +213,16 @@ export class HubE2E { }, "daemon to become connected"); } + private async sourceE2EDaemonSlug(): Promise { + const daemon = await this.requirePool().query<{ slug: string }>( + "select slug from daemons where id = $1", + [SOURCE_E2E_DAEMON_ID], + ); + const slug = daemon.rows[0]?.slug; + if (!slug) throw new Error("Source E2E daemon is unavailable"); + return slug; + } + async status(): Promise<{ state: string }> { const status = await this.requireSource().status(); return { state: requiredString(status, "state") }; @@ -348,12 +363,9 @@ export class HubE2E { async installProductionConfiguration( prompt = "Deploy requested for phase-five-operator", + workspaceAffinityKey?: string, ): Promise { - const daemon = await this.requirePool().query<{ slug: string }>( - "select slug from daemons where presence = 'connected' order by connected_at desc limit 1", - ); - const slug = daemon.rows[0]?.slug; - if (!slug) throw new Error("Connected daemon has no daemon slug"); + const slug = await this.sourceE2EDaemonSlug(); const yaml = [ "environments:", " - name: phase-five", @@ -372,6 +384,12 @@ export class HubE2E { " max_runtime: 1h", " idle_timeout: 5m", " auto_archive: true", + ...(workspaceAffinityKey === undefined + ? [] + : [ + " workspace_affinity:", + ` key: ${JSON.stringify(workspaceAffinityKey)}`, + ]), " agent:", " provider: hub-e2e", ` prompt: [{ text: ${JSON.stringify(prompt)} }]`, @@ -442,11 +460,7 @@ export class HubE2E { } async installRealAgentConfiguration(provider: RealAgentProvider): Promise { - const daemon = await this.requirePool().query<{ slug: string }>( - "select slug from daemons where presence = 'connected' order by connected_at desc limit 1", - ); - const slug = daemon.rows[0]?.slug; - if (!slug) throw new Error("Connected daemon has no daemon slug"); + const slug = await this.sourceE2EDaemonSlug(); const yaml = [ "environments:", " - name: real-agent", @@ -482,11 +496,7 @@ export class HubE2E { } async installRealAgentRoutingConfiguration(provider: RealAgentProvider): Promise { - const daemon = await this.requirePool().query<{ slug: string }>( - "select slug from daemons where presence = 'connected' order by connected_at desc limit 1", - ); - const slug = daemon.rows[0]?.slug; - if (!slug) throw new Error("Connected daemon has no daemon slug"); + const slug = await this.sourceE2EDaemonSlug(); const yaml = [ "environments:", " - name: real-agent-routing", @@ -985,6 +995,30 @@ export class HubE2E { }; } + async completedExecutionWorkspace(executionId: string): Promise { + await this.completedRun(executionId); + await this.observe(async () => { + const records = await this.persistedDaemonAgents(executionId); + return records.length === 1 && typeof records[0]?.archivedAt === "string"; + }, "completed execution agent archival"); + const record = (await this.persistedDaemonAgents(executionId))[0]; + if (!record?.workspaceId) throw new Error("Completed execution has no workspace"); + return record.workspaceId; + } + + async expectWorkspaceActive(workspaceId: string, active: boolean): Promise { + await this.observe( + async () => + (await this.requireSource().activeWorkspaceIds()).includes(workspaceId) === active, + `workspace ${workspaceId} to be ${active ? "active" : "archived"}`, + ); + } + + async archiveWorkspace(workspaceId: string): Promise { + await this.cli(["workspace", "archive", workspaceId, "--host", this.daemonHost, "--json"]); + await this.expectWorkspaceActive(workspaceId, false); + } + requestForbiddenOperation() { return this.requireProxy().requestForbiddenOperation(); } @@ -1365,7 +1399,7 @@ export class HubE2E { this.pool = await createPostgresQueryRuntime(this.postgres.getConnectionUri()); this.proxy = await HubFaultProxy.start(this.hubOrigin, proxyPort); this.hub = await this.startHub(); - if (this.options.realAgent !== true) { + if (this.options.realAgent !== true && this.options.completeInTurn !== true) { this.completionRunner = await startChild({ name: "completion-runner", command: process.execPath, @@ -1452,6 +1486,7 @@ export class HubE2E { HUB_E2E_ACP_RECORD_FILE: this.acpRecordFile, HUB_E2E_COMPLETE_GATE: this.completionGate, HUB_E2E_COMPLETION_JOBS: this.completionJobs, + HUB_E2E_COMPLETE_IN_TURN: this.options.completeInTurn === true ? "1" : "0", }, }, }, diff --git a/src/e2e/harness/source-paseo.lifecycle.test.ts b/src/e2e/harness/source-paseo.lifecycle.test.ts index eb3aa821..42bd753f 100644 --- a/src/e2e/harness/source-paseo.lifecycle.test.ts +++ b/src/e2e/harness/source-paseo.lifecycle.test.ts @@ -1,7 +1,40 @@ import assert from "node:assert/strict"; import { spawn } from "node:child_process"; import { describe, it } from "vitest"; -import { stopProcess } from "./source-paseo.js"; +import { sourceHubConnectArguments, stopProcess } from "./source-paseo.js"; + +describe("source Paseo Hub connection", () => { + it("keeps execution authority opt-in at the CLI boundary", () => { + const input = { + hubOrigin: "https://hub.test", + credential: "machine-key", + daemonHost: "127.0.0.1:4010", + }; + + assert.deepEqual(sourceHubConnectArguments(input), [ + "hub", + "connect", + "https://hub.test", + "--api-key", + "machine-key", + "--host", + "127.0.0.1:4010", + "--json", + ]); + assert.deepEqual(sourceHubConnectArguments({ ...input, permissions: ["hub.execute"] }), [ + "hub", + "connect", + "https://hub.test", + "--api-key", + "machine-key", + "--host", + "127.0.0.1:4010", + "--permission", + "hub.execute", + "--json", + ]); + }); +}); describe("source Paseo process ownership", () => { it("terminates a detached agent CLI child tree", async () => { diff --git a/src/e2e/harness/source-paseo.ts b/src/e2e/harness/source-paseo.ts index e01de654..01bebc0e 100644 --- a/src/e2e/harness/source-paseo.ts +++ b/src/e2e/harness/source-paseo.ts @@ -88,20 +88,17 @@ export class SourcePaseo { async connectWithCredential( hubOrigin: string, credential: string, + permissions: readonly string[] = [], ): Promise> { this.rememberedHubOrigin = hubOrigin; - const result = await this.run([ - "hub", - "connect", - hubOrigin, - "--api-key", - credential, - "--permission", - "hub.execute", - "--host", - this.paths.daemonHost, - "--json", - ]); + const result = await this.run( + sourceHubConnectArguments({ + hubOrigin, + credential, + daemonHost: this.paths.daemonHost, + permissions, + }), + ); await this.rememberActiveAuthority(); return result; } @@ -131,6 +128,19 @@ export class SourcePaseo { return z.array(z.unknown()).parse(value); } + async activeWorkspaceIds(): Promise { + const result = await runCommand( + join(this.paths.packagesRoot, "node_modules/.bin/paseo"), + ["workspace", "ls", "--host", this.paths.daemonHost, "--json"], + this.paths.packagesRoot, + sourceEnvironment(this.paths.paseoHome), + ); + return z + .array(z.object({ workspaceId: z.string() })) + .parse(JSON.parse(result.stdout)) + .map((workspace) => workspace.workspaceId); + } + async agentProvider(agentId: string): Promise { const result = await this.run([ "agent", @@ -282,6 +292,25 @@ export class SourcePaseo { } } +export function sourceHubConnectArguments(input: { + hubOrigin: string; + credential: string; + daemonHost: string; + permissions?: readonly string[]; +}): string[] { + return [ + "hub", + "connect", + input.hubOrigin, + "--api-key", + input.credential, + "--host", + input.daemonHost, + ...(input.permissions?.length ? ["--permission", ...input.permissions] : []), + "--json", + ]; +} + export function resolvePaseoWorktree(): string { const configured = process.env["PASEO_E2E_WORKTREE"]; if (!configured) { diff --git a/src/e2e/hub-workspace-affinity.e2e.test.ts b/src/e2e/hub-workspace-affinity.e2e.test.ts new file mode 100644 index 00000000..610a557e --- /dev/null +++ b/src/e2e/hub-workspace-affinity.e2e.test.ts @@ -0,0 +1,57 @@ +import assert from "node:assert/strict"; +import { afterEach, beforeAll, beforeEach, describe, it } from "vitest"; +import { HubE2E } from "./harness/index.js"; +import { prebuildPaseoArtifacts, resolvePaseoWorktree } from "./harness/source-paseo.js"; + +const describeAffinityE2E = process.env["RUN_HUB_AFFINITY_E2E"] === "1" ? describe : describe.skip; + +describeAffinityE2E("workspace affinity across daemon versions", () => { + let hub: HubE2E; + let affinitySupported: boolean; + + beforeAll(async () => { + const expectation = process.env["PASEO_E2E_AFFINITY_SUPPORTED"]; + assert.ok(expectation === "true" || expectation === "false", "declare daemon capability"); + affinitySupported = expectation === "true"; + await prebuildPaseoArtifacts(resolvePaseoWorktree()); + }, 600_000); + + beforeEach(async () => { + hub = await HubE2E.start({ completeInTurn: true }); + }, 120_000); + + afterEach(async () => { + const shutdown = await hub?.stop(); + assert.ok((shutdown?.durationMs ?? 0) < 10_000); + assert.deepEqual(shutdown?.leakedProcesses ?? [], []); + }, 120_000); + + it("preserves legacy behavior or reuses and restores the workspace when supported", async () => { + await hub.connect(); + await hub.daemonIsConnected(); + await hub.installProductionConfiguration(undefined, "conversation:one"); + + const first = await hub.runManual("affinity-first", "payments"); + const firstWorkspace = await hub.completedExecutionWorkspace(first.executionId); + await hub.expectWorkspaceActive(firstWorkspace, affinitySupported); + + const second = await hub.runManual("affinity-second", "payments"); + const secondWorkspace = await hub.completedExecutionWorkspace(second.executionId); + assert.notEqual(second.agentId, first.agentId); + assert.equal(secondWorkspace === firstWorkspace, affinitySupported); + await hub.expectWorkspaceActive(secondWorkspace, affinitySupported); + + if (affinitySupported) await hub.archiveWorkspace(secondWorkspace); + const restored = await hub.runManual("affinity-restored", "payments"); + const restoredWorkspace = await hub.completedExecutionWorkspace(restored.executionId); + assert.notEqual(restored.agentId, second.agentId); + assert.equal(restoredWorkspace === secondWorkspace, affinitySupported); + await hub.expectWorkspaceActive(restoredWorkspace, affinitySupported); + + await hub.installProductionConfiguration(undefined, "conversation:two"); + const unrelated = await hub.runManual("affinity-unrelated", "payments"); + const unrelatedWorkspace = await hub.completedExecutionWorkspace(unrelated.executionId); + assert.notEqual(unrelatedWorkspace, restoredWorkspace); + await hub.expectWorkspaceActive(unrelatedWorkspace, affinitySupported); + }, 180_000); +}); diff --git a/src/provider-applications/internal/runtime-owner.test.ts b/src/provider-applications/internal/runtime-owner.test.ts index 90d1b28e..0d5fd167 100644 --- a/src/provider-applications/internal/runtime-owner.test.ts +++ b/src/provider-applications/internal/runtime-owner.test.ts @@ -175,6 +175,7 @@ describe("dynamic provider runtime", () => { // The replaced source retires immediately; later callbacks use the active registration. assert.deepEqual(stopped, ["A1"]); + assert.equal(trigger.workspaceAffinityKey?.(oldMatch!.triggerContext), "A2:A1"); await trigger.onAgentExecutionCompleted?.(oldMatch!.triggerContext, oldMatch!.outputContext, { status: "succeeded", @@ -761,6 +762,7 @@ function fakeRegistration( completed.push(id); return Promise.resolve(); }, + workspaceAffinityKey: (triggerContext) => `${id}:${triggerContext.id}`, }; return { connection: { diff --git a/src/provider-applications/internal/runtime-owner.ts b/src/provider-applications/internal/runtime-owner.ts index 61bb42c5..335980ab 100644 --- a/src/provider-applications/internal/runtime-owner.ts +++ b/src/provider-applications/internal/runtime-owner.ts @@ -14,6 +14,7 @@ import type { } from "../../providers/registration.js"; import { createSlackRegistration } from "../../providers/slack/index.js"; import type { TriggerHandler, TriggerProvider, TriggerSource } from "../../triggers/index.js"; +import { providerSupportsWorkspaceAffinityConversationKey } from "../../triggers/workspace-affinity.js"; import type { Provider, ProviderApplicationConfiguration, @@ -483,6 +484,14 @@ export class DynamicProviderRuntime implements ProviderRuntimeOwner { const selected = current(); return this.withLease(selected.active, () => selected.trigger.match(external)); }, + ...(providerSupportsWorkspaceAffinityConversationKey(provider) + ? { + workspaceAffinityKey: (triggerContext: unknown) => { + const selected = current(); + return selected.trigger.workspaceAffinityKey?.(triggerContext); + }, + } + : {}), materializeLaunch: (input) => invoke((trigger) => trigger.materializeLaunch?.(input) ?? Promise.resolve({})), materializeContext: (input) => diff --git a/src/triggers/configuration/editor.test.ts b/src/triggers/configuration/editor.test.ts index 6ae4b886..6fbe4a3a 100644 --- a/src/triggers/configuration/editor.test.ts +++ b/src/triggers/configuration/editor.test.ts @@ -62,6 +62,11 @@ run: slack.reply: max: 3 auto_archive: false + continuation: + mode: new + workspace_affinity: + # preserve this identity exactly + key: " review-\${{ paseo.trigger.conversation_key }} " `; describe("trigger form YAML bridge", () => { @@ -107,6 +112,10 @@ describe("trigger form YAML bridge", () => { expect(value.run.output).toBeDefined(); expect(value.run.outputs).toEqual({ "slack.reply": { max: 3 } }); expect(value.run.auto_archive).toBe(false); + expect(value.run.workspace_affinity).toEqual({ + key: " review-${{ paseo.trigger.conversation_key }} ", + }); + expect(yaml).toContain("# preserve this identity exactly"); expect(value.on["slack.mention"]?.filters?.channels).toEqual(["engineering"]); }); @@ -123,6 +132,37 @@ describe("trigger form YAML bridge", () => { expect(value.run.agent.thinkingOptionId).toBe("xhigh"); }); + test("creates, changes, and explicitly removes affinity without normalizing keys", () => { + const projection = projectTriggerForm(ADVANCED); + if (projection.status !== "editable") throw new Error(projection.reason); + expect(projection.value.workspaceAffinityKey).toBe( + " review-${{ paseo.trigger.conversation_key }} ", + ); + const value = { ...projection.value, workspaceAffinityKey: " custom:key " }; + for (const yaml of [createTriggerYaml(value), patchTriggerYaml(ADVANCED, value)]) { + expect( + TriggerDocumentSchema.parse(parseDocument(yaml).toJS()).run.workspace_affinity, + ).toEqual({ + key: " custom:key ", + }); + const removed = patchTriggerYaml(yaml, { ...value, workspaceAffinityKey: "" }); + expect(removed).not.toContain("workspace_affinity"); + expect( + TriggerDocumentSchema.parse(parseDocument(removed).toJS()).run.workspace_affinity, + ).toBeUndefined(); + } + expect(patchTriggerYaml(ADVANCED, value)).toContain("# preserve this identity exactly"); + }); + + test("rejects whitespace-only affinity instead of silently disabling reuse", () => { + const projection = projectTriggerForm(ADVANCED); + if (projection.status !== "editable") throw new Error(projection.reason); + const value = { ...projection.value, workspaceAffinityKey: " " }; + expect(triggerFormErrors(value).workspaceAffinityKey).toMatch(/nonblank affinity key/u); + expect(() => createTriggerYaml(value)).toThrow(/nonblank affinity key/u); + expect(() => patchTriggerYaml(ADVANCED, value)).toThrow(/nonblank affinity key/u); + }); + test("keeps advanced YAML added before a new trigger's first save", () => { const projection = projectTriggerForm(ADVANCED); if (projection.status !== "editable") throw new Error(projection.reason); @@ -168,6 +208,7 @@ describe("trigger form YAML bridge", () => { continuationKey: "", maxRuntime: "2h", idleTimeout: "10m", + workspaceAffinityKey: "", githubConnection: "", githubRepositories: "", githubPermissions: "", @@ -207,6 +248,7 @@ describe("trigger form YAML bridge", () => { continuationKey: "", maxRuntime: "2h", idleTimeout: "10m", + workspaceAffinityKey: "", githubConnection: "", githubRepositories: "", githubPermissions: "", @@ -437,8 +479,8 @@ test.each(["github.issue_label_added", "github.pull_request_label_added"])( test("continuation policies round-trip through the form and YAML", () => { const projection = projectTriggerForm(ADVANCED); if (projection.status !== "editable") throw new Error(projection.reason); - expect(projection.value.continuationMode).toBe("conversation"); - for (const mode of ["key", "new", "conversation"]) { + expect(projection.value.continuationMode).toBe("new"); + for (const mode of ["new"]) { const yaml = patchTriggerYaml(ADVANCED, { ...projection.value, continuationMode: mode, @@ -449,3 +491,17 @@ test("continuation policies round-trip through the form and YAML", () => { expect(yaml).toContain("# keep this heading"); } }); + +test("rejects agent continuation when workspace affinity is enabled", () => { + const projection = projectTriggerForm(ADVANCED); + if (projection.status !== "editable") throw new Error(projection.reason); + for (const mode of ["conversation", "key"]) { + const value = { + ...projection.value, + continuationMode: mode, + continuationKey: "shared-agent", + }; + expect(triggerFormErrors(value).continuationMode).toMatch(/New agent continuity/u); + expect(() => patchTriggerYaml(ADVANCED, value)).toThrow(/New agent continuity/u); + } +}); diff --git a/src/triggers/configuration/editor.ts b/src/triggers/configuration/editor.ts index 42f7dd40..d0fb2242 100644 --- a/src/triggers/configuration/editor.ts +++ b/src/triggers/configuration/editor.ts @@ -8,6 +8,7 @@ import { ContinuationSchema } from "../continuation.js"; import { parseDocument, stringify, type Document } from "yaml"; import { z } from "zod"; import { IDENTIFIER, TriggerDocumentSchema, type TriggerDocument } from "./schema.js"; +import { WorkspaceAffinityKeySchema } from "../../config/workspace-affinity.js"; import { eventDefinition, @@ -35,6 +36,7 @@ export interface TriggerFormValue { continuationKey: string; maxRuntime: string; idleTimeout: string; + workspaceAffinityKey: string; githubConnection: string; githubRepositories: string; githubPermissions: string; @@ -84,7 +86,7 @@ function toFormValue( connection: definition.connection ?? "", allowedUsers: definition.filters?.from_users?.join(", ") ?? "*", qualifiers: readQualifiers(event, definition.filters), - recurrence: definition.recurrence ?? DEFAULT_RECURRENCE, + recurrence: formRecurrence(definition), daemon: trigger.run.target.daemon, cwd: trigger.run.target.cwd, agent: joinAgentId(agent.provider, agent.model), @@ -95,6 +97,7 @@ function toFormValue( continuationKey: trigger.run.continuation.mode === "key" ? trigger.run.continuation.key : "", maxRuntime: trigger.run.max_runtime, idleTimeout: trigger.run.idle_timeout, + workspaceAffinityKey: trigger.run.workspace_affinity?.key ?? "", githubConnection: trigger.run.github?.connection ?? "", githubRepositories: trigger.run.github?.repositories?.join(", ") ?? "", githubPermissions: @@ -106,6 +109,10 @@ function toFormValue( }; } +function formRecurrence(definition: TriggerDocument["on"][string]): Recurrence { + return definition.recurrence ?? DEFAULT_RECURRENCE; +} + /** Patch only form-owned YAML nodes, retaining comments, ordering, and advanced nodes. */ export function patchTriggerYaml(yaml: string, value: TriggerFormValue): string { const projection = projectTriggerForm(yaml); @@ -163,6 +170,8 @@ export function patchTriggerYaml(yaml: string, value: TriggerFormValue): string setIfChanged(document, ["run", "continuation"], formContinuation(value)); setIfChanged(document, ["run", "max_runtime"], value.maxRuntime.trim()); setIfChanged(document, ["run", "idle_timeout"], value.idleTimeout.trim()); + if (value.workspaceAffinityKey === "") deleteIfPresent(document, ["run", "workspace_affinity"]); + else setIfChanged(document, ["run", "workspace_affinity", "key"], value.workspaceAffinityKey); setOptional(document, ["run", "github"], githubAuthority(value)); setIfChanged(document, ["run", "prompt"], value.prompt); return document.toString({ lineWidth: 0 }); @@ -226,6 +235,9 @@ export function createTriggerYaml(value: TriggerFormValue): string { continuation: formContinuation(value), max_runtime: value.maxRuntime.trim(), idle_timeout: value.idleTimeout.trim(), + ...(value.workspaceAffinityKey === "" + ? {} + : { workspace_affinity: { key: value.workspaceAffinityKey } }), ...(github === undefined ? {} : { github }), prompt: value.prompt, }, @@ -305,12 +317,7 @@ export function triggerFormErrors(value: TriggerFormValue): TriggerFieldErrors { errors.allowedUsers = "Name at least one user ID, or let everyone trigger it."; } } - for (const qualifier of eventDefinition(value.event).qualifiers) { - const selection = value.qualifiers[qualifier.key]; - if (qualifier.required && (selection === undefined || selection.trim().length === 0)) { - errors[`qualifiers.${qualifier.key}`] = `${qualifier.label} is required.`; - } - } + Object.assign(errors, requiredQualifierErrors(value)); Object.assign(errors, recurrenceErrors(value)); if (value.daemon.trim().length === 0) errors.daemon = "Daemon is required."; if (!value.cwd.trim().startsWith("/")) { @@ -318,6 +325,15 @@ export function triggerFormErrors(value: TriggerFormValue): TriggerFieldErrors { } if (value.maxRuntime.trim().length === 0) errors.maxRuntime = "Maximum runtime is required."; if (value.idleTimeout.trim().length === 0) errors.idleTimeout = "Idle timeout is required."; + if ( + value.workspaceAffinityKey !== "" && + !WorkspaceAffinityKeySchema.safeParse(value.workspaceAffinityKey).success + ) { + errors.workspaceAffinityKey = "Enter a nonblank affinity key, or clear it to disable reuse."; + } + if (value.workspaceAffinityKey !== "" && value.continuationMode !== "new") { + errors.continuationMode = "Workspace affinity requires New agent continuity."; + } const agent = refused(() => splitAgentId(value.agent)); if (agent !== undefined) errors.agent = agent; if (value.mode.trim().length === 0) errors.mode = "Execution mode is required."; @@ -332,6 +348,17 @@ export function triggerFormErrors(value: TriggerFormValue): TriggerFieldErrors { return errors; } +function requiredQualifierErrors(value: TriggerFormValue): TriggerFieldErrors { + const errors: TriggerFieldErrors = {}; + for (const qualifier of eventDefinition(value.event).qualifiers) { + const selection = value.qualifiers[qualifier.key]; + if (qualifier.required && (selection === undefined || selection.trim().length === 0)) { + errors[`qualifiers.${qualifier.key}`] = `${qualifier.label} is required.`; + } + } + return errors; +} + /** The message a parse refused with, or `undefined` when it accepted the value. */ function refused(parse: () => unknown): string | undefined { try { diff --git a/src/triggers/configuration/index.test.ts b/src/triggers/configuration/index.test.ts index 09dd84cd..0a12db9e 100644 --- a/src/triggers/configuration/index.test.ts +++ b/src/triggers/configuration/index.test.ts @@ -121,6 +121,67 @@ describe("self-contained trigger documents", () => { assert.deepEqual(parseTriggerDocument(serializeTriggerDocument(parsed)), parsed); }); + it.each([ + " shared-review ", + "review:${{ paseo.trigger.conversation_key }}", + "review:${{ paseo.inputs.model }}", + ])( + "preserves affinity %s for every event and keeps workspace retention separate from idle timeout", + (key) => { + const document = parseTriggerDocument(trigger); + document.max_runtime = "4h"; + document.run.continuation = { mode: "new" }; + document.run.workspace_affinity = { key }; + document.run.auto_archive = false; + const yaml = serializeTriggerDocument(document); + const compiled = compileTriggerDocument(yaml); + assert.deepEqual(parseTriggerDocument(yaml).run.workspace_affinity, { key }); + for (const event of compiled.events) { + assert.equal(event.maxRuntimeMs, 4 * 60 * 60_000); + assert.deepEqual(event.steps[0]?.workspaceAffinity, { key }); + assert.equal(event.steps[0]?.maxRuntimeMs, 90 * 60_000); + assert.equal(event.steps[0]?.idleTimeoutMs, 10 * 60_000); + assert.equal(event.steps[0]?.autoArchive, false); + } + }, + ); + + it.each([ + "${{ paseo.prompt }}", + "${{ paseo.context.linear.issue.id }}", + "${{ paseo.execution.id }}", + ])("rejects untrusted workspace selection %s through the single-run compiler", (key) => { + const document = parseTriggerDocument(trigger); + document.run.continuation = { mode: "new" }; + document.run.workspace_affinity = { key }; + assert.throws( + () => compileTriggerDocument(serializeTriggerDocument(document)), + TriggerDocumentError, + ); + }); + + it("validates the conversation key against every subscribed event", () => { + const document = parseTriggerDocument(trigger); + document.on["manual.run"] = {}; + document.run.continuation = { mode: "new" }; + document.run.workspace_affinity = { key: "${{ paseo.trigger.conversation_key }}" }; + assert.throws( + () => compileTriggerDocument(serializeTriggerDocument(document)), + /manual\.run does not provide a conversation key/u, + ); + }); + + it("rejects an execution-scoped worktree for a single-run affinity key", () => { + const document = parseTriggerDocument(trigger); + document.run.continuation = { mode: "new" }; + document.run.workspace_affinity = { key: "shared-review" }; + document.run.target.worktree = { + mode: "branch-off", + newBranch: "run-${{ paseo.execution.id }}", + }; + assert.throws(() => compileTriggerDocument(serializeTriggerDocument(document)), /execution/u); + }); + it("allows authenticated manual dispatches when no actor filter is authored", () => { const compiled = compileTriggerDocument(` name: deploy diff --git a/src/triggers/configuration/index.ts b/src/triggers/configuration/index.ts index fd2de085..8900e1c5 100644 --- a/src/triggers/configuration/index.ts +++ b/src/triggers/configuration/index.ts @@ -74,6 +74,9 @@ export function compileTriggerDocument(yaml: string): CompiledTriggerDocument { ...(authored.run.output === undefined ? {} : { output: authored.run.output }), ...(allowOutputs.length === 0 ? {} : { allow_outputs: allowOutputs }), auto_archive: authored.run.auto_archive, + ...(authored.run.workspace_affinity === undefined + ? {} + : { workspace_affinity: authored.run.workspace_affinity }), }, ], }; diff --git a/src/triggers/configuration/legacy-migration.test.ts b/src/triggers/configuration/legacy-migration.test.ts index 1097025a..48ef29b8 100644 --- a/src/triggers/configuration/legacy-migration.test.ts +++ b/src/triggers/configuration/legacy-migration.test.ts @@ -58,6 +58,48 @@ steps: assert.doesNotMatch(trigger.yaml, /include:|partials|steps:/u); }); + it.each([" custom-workspace ", "slack:${{ paseo.trigger.conversation_key }}"])( + "preserves single-step affinity %s in the current trigger format", + (key) => { + const migrated = migrateLegacyBundle({ + files: [ + { path: ".paseo/hub.yml", content: hub }, + { + path: ".paseo/workflows/affinity.yml", + content: ` +name: affinity +on: slack.mention +filters: { from_users: [U123] } +max_runtime: 2h +steps: + - id: work + environment: runner + max_runtime: 90m + idle_timeout: 10m + auto_archive: false + workspace_affinity: { key: ${JSON.stringify(key)} } + agent: codex + prompt: [{ text: work }] +`, + }, + ], + }); + + assert.equal(migrated.length, 1); + const trigger = migrated[0]; + assert.equal(trigger?.format, "single_run"); + if (trigger?.format !== "single_run") return; + assert.deepEqual(trigger.compiled.authored.run.workspace_affinity, { key }); + const event = trigger.compiled.events[0]!; + assert.deepEqual(event.steps[0]?.workspaceAffinity, { key }); + assert.equal(event.maxRuntimeMs, 2 * 60 * 60_000); + assert.equal(event.steps[0]?.maxRuntimeMs, 90 * 60_000); + assert.equal(event.steps[0]?.idleTimeoutMs, 10 * 60_000); + assert.equal(event.steps[0]?.autoArchive, false); + assert.match(trigger.yaml, /workspace_affinity:/u); + }, + ); + it("preserves a multi-step workflow as one self-contained normalized legacy trigger", () => { const migrated = migrateLegacyBundle({ files: [ diff --git a/src/triggers/configuration/legacy-migration.ts b/src/triggers/configuration/legacy-migration.ts index d8ec5941..dc09aa7d 100644 --- a/src/triggers/configuration/legacy-migration.ts +++ b/src/triggers/configuration/legacy-migration.ts @@ -176,6 +176,9 @@ function singleRunDocument( : { output: { schema: structuredClone(asJsonObject(step.output.schema)) } }), ...(Object.keys(outputs).length === 0 ? {} : { outputs }), auto_archive: step.autoArchive, + ...(step.workspaceAffinity === undefined + ? {} + : { workspace_affinity: { key: step.workspaceAffinity.key } }), }, }; } diff --git a/src/triggers/configuration/schema.ts b/src/triggers/configuration/schema.ts index c19decc6..1184e067 100644 --- a/src/triggers/configuration/schema.ts +++ b/src/triggers/configuration/schema.ts @@ -3,6 +3,7 @@ import { z } from "zod"; import { ContinuationSchema } from "../continuation.js"; import { eventDefinition, isEditorEvent } from "./events.js"; import { AuthoredGitHubAuthoritySchema } from "../../config/github-authority.js"; +import { WorkspaceAffinitySchema } from "../../config/workspace-affinity.js"; type JsonPrimitive = string | number | boolean | null; type JsonValue = JsonPrimitive | JsonValue[] | { [key: string]: JsonValue }; @@ -111,6 +112,7 @@ export const TriggerRunSchema = z .optional(), outputs: z.record(z.string().regex(EVENT_NAME), TriggerOutputSchema).optional(), auto_archive: z.boolean().default(true), + workspace_affinity: WorkspaceAffinitySchema.optional(), }) .strict(); @@ -179,6 +181,13 @@ export const TriggerDocumentSchema = z message: "at least one agent choice is required", }); } + if (trigger.run.workspace_affinity !== undefined && trigger.run.continuation.mode !== "new") { + context.addIssue({ + code: z.ZodIssueCode.custom, + path: ["run", "continuation"], + message: "Workspace affinity requires New agent continuity.", + }); + } }); export type TriggerDocument = z.infer; diff --git a/src/triggers/discord/provider.test.ts b/src/triggers/discord/provider.test.ts index ac0b336b..0af160d3 100644 --- a/src/triggers/discord/provider.test.ts +++ b/src/triggers/discord/provider.test.ts @@ -133,6 +133,41 @@ describe("Discord Phase 1 trigger provider", () => { ); }); + it("uses one authenticated affinity key for a Discord thread starter and its replies", async () => { + const { project, revision, store } = await activeConfiguration(); + const provider = createDiscordTriggerProvider({ + configurationStoreForProject: () => store, + bot: new MemoryDiscordBotClient({ selfUserId: "900" }), + }); + const starter = ( + await provider.match( + external(project.id, revision.id, event({ channelId: "200", messageId: "300" })), + ) + )[0]; + const reply = ( + await provider.match( + external( + project.id, + revision.id, + event({ + channelId: "300", + threadId: "300", + parentChannelId: "200", + messageId: "301", + }), + ), + ) + )[0]; + if (!isAcceptedTriggerProviderMatch(starter) || !isAcceptedTriggerProviderMatch(reply)) { + throw new Error("expected accepted matches"); + } + + assert.equal( + provider.workspaceAffinityKey?.(starter.triggerContext), + provider.workspaceAffinityKey?.(reply.triggerContext), + ); + }); + it("routes a durable Discord receipt to the configured connection", async () => { const database = createMemoryDatabase(); const connection = { diff --git a/src/triggers/discord/provider.ts b/src/triggers/discord/provider.ts index 77ced095..ff320365 100644 --- a/src/triggers/discord/provider.ts +++ b/src/triggers/discord/provider.ts @@ -217,6 +217,17 @@ export function createDiscordTriggerProvider(options: { }, }; }, + workspaceAffinityKey(triggerContext) { + const target = triggerContext.target; + const thread = triggerContext.event.discord.trigger_message.thread; + return JSON.stringify([ + "discord", + triggerContext.event.discord.connection_id, + target.guildId, + thread?.parent_channel_id ?? target.channelId, + thread?.id ?? target.messageId, + ]); + }, async onDispatchAccepted(triggerContext, _outputContext, reactionState) { if (discordReactionPhase(reactionState) !== undefined) return reactionState; await reactSafely(options.bot, triggerContext.target, "eyes"); diff --git a/src/triggers/github/provider.test.ts b/src/triggers/github/provider.test.ts index 617cb1c2..882d3c96 100644 --- a/src/triggers/github/provider.test.ts +++ b/src/triggers/github/provider.test.ts @@ -85,6 +85,26 @@ describe("GitHub Phase 1 trigger provider", () => { assert.equal(match.invocation.prompt, prompt); }); + it("derives a stable authenticated affinity key for comments on the same GitHub item", async () => { + const { project, revision, store } = await activeConfiguration(); + const provider = createProvider(store, new TestReactions()); + const first = (await provider.match(external(project.id, revision.id, createEvent())))[0]; + const secondEvent = { ...createEvent({ body: "follow up @paseo" }), id: "github-delivery-2" }; + const second = (await provider.match(external(project.id, revision.id, secondEvent)))[0]; + if (!isAcceptedTriggerProviderMatch(first) || !isAcceptedTriggerProviderMatch(second)) { + throw new Error("expected accepted matches"); + } + + assert.equal( + provider.workspaceAffinityKey?.(first.triggerContext), + JSON.stringify(["github", null, 7, "issue", 211]), + ); + assert.equal( + provider.workspaceAffinityKey?.(first.triggerContext), + provider.workspaceAffinityKey?.(second.triggerContext), + ); + }); + it("matches a literal one-step prompt only after the security filters pass", async () => { const { project, revision, store } = await activeConfiguration(); const reactions = new TestReactions(); diff --git a/src/triggers/github/provider.ts b/src/triggers/github/provider.ts index b0f4ed55..efdbaff8 100644 --- a/src/triggers/github/provider.ts +++ b/src/triggers/github/provider.ts @@ -128,7 +128,12 @@ export type GitHubReactionSubject = export interface GitHubTriggerContext { provider: "github"; - target: { installationId: number; repository: string }; + target: { + installationId: number; + repository: string; + repositoryId: number; + connectionId: string | null; + }; event: GitHubMergeData; reactionSubject: GitHubReactionSubject | null; } @@ -171,7 +176,12 @@ export function createGitHubTriggerProvider(options: { throw new Error(`compiled trigger not found: ${match.trigger.name}`); const triggerContext: GitHubTriggerContext = { provider: "github", - target: { installationId: event.installationId, repository: event.repo }, + target: { + installationId: event.installationId, + repository: event.repo, + repositoryId: event.repositoryId, + connectionId: externalTrigger.connectionId ?? null, + }, event: buildGitHubMergeData(event), reactionSubject: reactionSubjectForEvent(event), }; @@ -212,6 +222,17 @@ export function createGitHubTriggerProvider(options: { async materializeContext(launch) { return launch.triggerContext.event; }, + workspaceAffinityKey(triggerContext) { + const item = triggerContext.event.github.item; + if (item === null || item.number === null) return undefined; + return JSON.stringify([ + "github", + triggerContext.target.connectionId, + triggerContext.target.repositoryId, + item.type, + item.number, + ]); + }, async onDispatchAccepted(triggerContext, _outputContext, reactionState) { if (triggerContext.reactionSubject === null) return null; if (githubReactionId(reactionState) !== undefined) return reactionState; diff --git a/src/triggers/index.ts b/src/triggers/index.ts index 0c687376..2a5e9873 100644 --- a/src/triggers/index.ts +++ b/src/triggers/index.ts @@ -144,6 +144,11 @@ export interface TriggerProvider< materializeContext?( launch: TriggerContextMaterialization, ): Promise; + /** + * Returns a provider-authenticated conversation identity for workspace affinity. It must never + * be derived from untrusted prompt text. + */ + workspaceAffinityKey?(triggerContext: TriggerContext): string | undefined; onDispatchAccepted?( triggerContext: TriggerContext, outputContext: OutputContext, diff --git a/src/triggers/linear/provider.test.ts b/src/triggers/linear/provider.test.ts index 5ac1cf2f..0ea64cf6 100644 --- a/src/triggers/linear/provider.test.ts +++ b/src/triggers/linear/provider.test.ts @@ -8,6 +8,74 @@ import type { NormalizedLinearCommentEvent } from "./events.js"; import { createLinearTriggerProvider } from "./provider.js"; describe("Linear trigger provider", () => { + it("uses the same issue identity across comments and issue events, isolated by connection and tenant", async () => { + const configuration = linearCommentConfiguration(); + const comment = configuration.triggers[0]!; + const { project, revision, store } = await activeConfiguration({ + ...configuration, + triggers: [comment, { ...comment, name: "issue", on: "linear.issue_entered_scope" }], + }); + const provider = createLinearTriggerProvider({ configurationStoreForProject: () => store }); + const original = external(project.id, revision.id); + const payload = event("2026-01-02T00:00:00.000Z"); + assert.ok(payload.issue); + const keyFor = async (input: ExternalTrigger) => { + const matches = await provider.match(input); + if (typeof matches === "string" || !isAcceptedTriggerProviderMatch(matches[0])) { + throw new Error("expected accepted Linear match"); + } + return provider.workspaceAffinityKey?.(matches[0].triggerContext); + }; + const key = await keyFor(original); + assert.equal(key, JSON.stringify(["linear", "linear-connection", "linear-org", "issue-1"])); + assert.equal( + await keyFor({ + ...original, + deliveryId: "another-delivery", + payload: { + ...payload, + id: "comment-2", + comment: { ...payload.comment, id: "comment-2", body: "another request" }, + issue: { ...payload.issue, identifier: "MOVED-99", title: "Renamed" }, + }, + }), + key, + ); + assert.equal( + await keyFor({ + ...original, + source: "linear.issue", + payload: { + type: "issue", + action: "create", + id: payload.issue.id, + organizationId: payload.organizationId, + actor: payload.actor, + issue: payload.issue, + updatedFrom: {}, + }, + }), + key, + ); + assert.notEqual(await keyFor({ ...original, connectionId: "another-connection" }), key); + assert.notEqual(await keyFor({ ...original, connectionId: null }), key); + assert.notEqual( + await keyFor({ ...original, payload: { ...payload, organizationId: "another-tenant" } }), + key, + ); + assert.notEqual( + await keyFor({ + ...original, + payload: { + ...payload, + comment: { ...payload.comment, issueId: "issue-2" }, + issue: { ...payload.issue, id: "issue-2" }, + }, + }), + key, + ); + }); + it.each([ ["pattern", { pattern: "/run" }, "/run priority=high investigate"], ["contains", { contains: "/run" }, "please /run priority=high investigate"], diff --git a/src/triggers/linear/provider.ts b/src/triggers/linear/provider.ts index ce1d3bf6..a0ed0018 100644 --- a/src/triggers/linear/provider.ts +++ b/src/triggers/linear/provider.ts @@ -160,6 +160,16 @@ export function createLinearTriggerProvider(options: { } return matches.length === 0 ? "trigger_filters_rejected" : matches; }, + workspaceAffinityKey(triggerContext) { + // The issue owns the files. Comments, deliveries, and (in the session integration) agent + // sessions own individual turns/replies, not workspace identity. Use UUIDs, not ENG-123. + return JSON.stringify([ + "linear", + triggerContext.event.linear.connection_id, + triggerContext.target.linearOrganizationId, + triggerContext.target.issueId, + ]); + }, async materializeContext(launch): Promise { const { trigger_thread_context: locator, ...linear } = launch.triggerContext.event.linear; const root = issueRootMessage(linear.issue); diff --git a/src/triggers/migration.test.ts b/src/triggers/migration.test.ts index 3cd76d46..4770ab56 100644 --- a/src/triggers/migration.test.ts +++ b/src/triggers/migration.test.ts @@ -1,6 +1,8 @@ import assert from "node:assert/strict"; import { describe, it } from "vitest"; import { compileHubBundle, type HubBundleFile } from "../config/bundle.js"; +import { parseCompiledHubConfig } from "../config/compiler.js"; +import { compileTriggerDocument } from "./configuration/index.js"; import { createMemoryDatabase } from "../db/memory.js"; import type { Database, ProjectRecord } from "../db/types.js"; import { migrateLegacyProjectTriggers } from "./migration.js"; @@ -71,6 +73,55 @@ describe("startup project trigger migration", () => { assert.equal((await database.listOrganizationTriggers("org")).length, 2); }); + it("preserves a single-step affinity configuration durably across startup migration", async () => { + const database = createMemoryDatabase({ organizationIds: ["org"] }); + const key = "slack:${{ paseo.trigger.conversation_key }}"; + await activeProject(database, "affinity", [ + workflow( + "affinity", + "slack.mention", + `${oneStep()} workspace_affinity: { key: ${JSON.stringify(key)} }\n`, + ), + ]); + const source = (await database.listPendingProjectTriggerMigrations())[0]!; + const original = parseCompiledHubConfig(source.revision.normalizedConfiguration).triggers[0]!; + + assert.deepEqual(await migrateLegacyProjectTriggers(database), { + projects: 1, + triggers: 1, + legacyMultistepTriggers: 0, + }); + const [trigger] = await database.listOrganizationTriggers("org"); + assert.equal(trigger?.format, "single_run"); + assert.ok(trigger); + const revision = await database.findOrganizationTriggerRevision( + trigger.id, + trigger.activeRevisionId, + ); + assert.ok(revision); + const migrated = parseCompiledHubConfig(revision.normalizedConfiguration).triggers[0]!; + assert.deepEqual(migrated.steps[0]?.workspaceAffinity, { key }); + assert.equal(migrated.maxRuntimeMs, original.maxRuntimeMs); + assert.equal(migrated.steps[0]?.maxRuntimeMs, original.steps[0]?.maxRuntimeMs); + assert.equal(migrated.steps[0]?.idleTimeoutMs, original.steps[0]?.idleTimeoutMs); + assert.equal(migrated.steps[0]?.autoArchive, original.steps[0]?.autoArchive); + assert.deepEqual(compileTriggerDocument(revision.yaml).events[0]?.steps[0]?.workspaceAffinity, { + key, + }); + assert.match(revision.yaml, /workspace_affinity:/u); + + assert.deepEqual(await migrateLegacyProjectTriggers(database), { + projects: 0, + triggers: 0, + legacyMultistepTriggers: 0, + }); + assert.deepEqual(await database.listOrganizationTriggers("org"), [trigger]); + assert.deepEqual( + await database.findOrganizationTriggerRevision(trigger.id, trigger.activeRevisionId), + revision, + ); + }); + it("deterministically disambiguates duplicate trigger names when projects collapse", async () => { const database = createMemoryDatabase({ organizationIds: ["org"] }); await activeProject(database, "hub", [workflow("request", "manual.run", oneStep())]); diff --git a/src/triggers/panel.tsx b/src/triggers/panel.tsx index ecf3aa8b..50ce94da 100644 --- a/src/triggers/panel.tsx +++ b/src/triggers/panel.tsx @@ -44,6 +44,7 @@ import type { HubProviderSnapshot, HubProviderSnapshotEntry } from "../hub/proto import { EventFields } from "./event-fields.js"; import { EDITOR_EVENTS, eventDefinition, parseEditorEvent } from "./configuration/events.js"; import { selectedProviderModel } from "./provider-catalog.js"; +import { triggerSupportsWorkspaceAffinityConversationKey } from "./workspace-affinity.js"; type BrowserTrigger = TriggerSnapshot["triggers"][number]; type EditorMode = "form" | "yaml"; @@ -596,6 +597,7 @@ function TriggerForm({ const githubEnabled = form.githubConnection !== ""; const [githubExpanded, setGithubExpanded] = useState(githubEnabled); const [optionsExpanded, setOptionsExpanded] = useState(false); + const [affinityExpanded, setAffinityExpanded] = useState(form.workspaceAffinityKey !== ""); const everyone = form.allowedUsers.trim() === "*"; const update = (key: Key, value: TriggerFormValue[Key]) => onChange({ ...form, [key]: value }); @@ -799,6 +801,41 @@ function TriggerForm({ /> )} + {selectedDaemon === undefined ? null : ( + + + + Each arrival creates a separate agent; runs can overlap in the same files. + Supporting daemons retain the workspace through the latest trigger maximum + runtime, including gaps between runs. Idle timeout only stops unresponsive + agents. Older daemons ignore affinity and create fresh workspaces. + + + )} )} @@ -1280,6 +1317,7 @@ function defaultForm(snapshot: TriggerSnapshot): TriggerFormValue { continuationKey: "", maxRuntime: "2h", idleTimeout: "10m", + workspaceAffinityKey: "", githubConnection: "", githubRepositories: "", githubPermissions: "", diff --git a/src/triggers/slack/provider.test.ts b/src/triggers/slack/provider.test.ts index 2ca94fe5..74277c09 100644 --- a/src/triggers/slack/provider.test.ts +++ b/src/triggers/slack/provider.test.ts @@ -240,6 +240,41 @@ describe("Slack Phase 1 trigger provider", () => { ); }); + it("uses one authenticated affinity key for a Slack root and its replies", async () => { + const database = createMemoryDatabase(); + const { project, revision, store } = await createActiveProjectConfiguration( + database, + configuration(), + { organizationId: "org-1" }, + ); + const provider = createSlackTriggerProvider({ + configurationStoreForProject: () => store, + botUserIdForWorkspace: () => Promise.resolve("UBOT"), + client: new RecordingSlackClient(), + }); + const root = ( + await provider.match( + external(project.id, revision.id, { threadTs: null, messageTs: "1700000000.000001" }), + ) + )[0]; + const reply = ( + await provider.match( + external(project.id, revision.id, { + threadTs: "1700000000.000001", + messageTs: "1700000001.000001", + }), + ) + )[0]; + if (!isAcceptedTriggerProviderMatch(root) || !isAcceptedTriggerProviderMatch(reply)) { + throw new Error("expected accepted matches"); + } + + assert.equal( + provider.workspaceAffinityKey?.(root.triggerContext), + provider.workspaceAffinityKey?.(reply.triggerContext), + ); + }); + it("targets Slack failure output at the originating message thread", async () => { const database = createMemoryDatabase(); const { project, revision, store } = await createActiveProjectConfiguration( diff --git a/src/triggers/slack/provider.ts b/src/triggers/slack/provider.ts index d26dc3fd..3ed5fd53 100644 --- a/src/triggers/slack/provider.ts +++ b/src/triggers/slack/provider.ts @@ -256,6 +256,16 @@ export function createSlackTriggerProvider(options: { }, }; }, + workspaceAffinityKey(triggerContext) { + const target = triggerContext.target; + return JSON.stringify([ + "slack", + triggerContext.event.slack.connection_id, + target.teamId, + target.channelId, + target.threadTs, + ]); + }, async onDispatchAccepted(_triggerContext, _outputContext, reactionState) { if (slackReactionPhase(reactionState) !== undefined) return reactionState; return { phase: "accepted" }; diff --git a/src/triggers/workspace-affinity.test.ts b/src/triggers/workspace-affinity.test.ts new file mode 100644 index 00000000..99237fbf --- /dev/null +++ b/src/triggers/workspace-affinity.test.ts @@ -0,0 +1,40 @@ +import { describe, expect, it } from "vitest"; +import { + providerSupportsWorkspaceAffinityConversationKey, + triggerSupportsWorkspaceAffinityConversationKey, +} from "./workspace-affinity.js"; +import { compileTriggerDocument } from "./configuration/index.js"; + +describe("workspace conversation identity support", () => { + it.each(["linear.issue_entered_scope", "linear.issue_assigned", "linear.comment_created"])( + "compiles an issue-scoped key for %s", + (event) => { + expect(providerSupportsWorkspaceAffinityConversationKey("linear")).toBe(true); + expect(triggerSupportsWorkspaceAffinityConversationKey(event)).toBe(true); + const compiled = compileTriggerDocument(` +name: linear-work +on: + ${event}: + connection: company-linear + filters: { from_users: [operator], project: project-1 } +run: + target: { daemon: devbox, cwd: /repo } + agent: { provider: codex } + continuation: { mode: new } + prompt: Handle the request + workspace_affinity: + key: "\${{ paseo.trigger.conversation_key }}" +`); + expect(compiled.events[0]?.steps[0]?.workspaceAffinity).toEqual({ + key: "${{ paseo.trigger.conversation_key }}", + }); + }, + ); + + it.each(["manual.run", "github.push", "linear.unknown", "linear.agent_session_created"])( + "does not promise a conversation identity before an event is implemented: %s", + (event) => { + expect(triggerSupportsWorkspaceAffinityConversationKey(event)).toBe(false); + }, + ); +}); diff --git a/src/triggers/workspace-affinity.ts b/src/triggers/workspace-affinity.ts new file mode 100644 index 00000000..d9d53bfe --- /dev/null +++ b/src/triggers/workspace-affinity.ts @@ -0,0 +1,27 @@ +const CONVERSATION_KEY_PROVIDERS = new Set(["github", "slack", "discord", "linear"]); +const CONVERSATION_KEY_TRIGGER_EVENTS = new Set([ + "github.issue_comment", + "github.issues", + "github.pull_request", + "github.pull_request_review", + "github.pull_request_review_comment", + "github.issue_created", + "github.pull_request_created", + "github.issue_comment_created", + "github.pull_request_comment_created", + "github.issue_label_added", + "github.pull_request_label_added", + "slack.mention", + "discord.mention", + "linear.issue_entered_scope", + "linear.issue_assigned", + "linear.comment_created", +]); + +export function providerSupportsWorkspaceAffinityConversationKey(provider: string): boolean { + return CONVERSATION_KEY_PROVIDERS.has(provider); +} + +export function triggerSupportsWorkspaceAffinityConversationKey(eventName: string): boolean { + return CONVERSATION_KEY_TRIGGER_EVENTS.has(eventName); +} diff --git a/src/workflows/engine.test.ts b/src/workflows/engine.test.ts index c212d35f..16473156 100644 --- a/src/workflows/engine.test.ts +++ b/src/workflows/engine.test.ts @@ -19,7 +19,7 @@ import type { ProviderEventReceiptRecord, TriggerRunRecord, } from "../db/types.js"; -import type { AcceptedTriggerProviderMatch } from "../triggers/index.js"; +import type { AcceptedTriggerProviderMatch, TriggerEventName } from "../triggers/index.js"; import type { LaunchMachineIntent } from "../dispatcher/launch-machine-intent.js"; import { parseInvocation } from "../triggers/invocation.js"; import { UNLIMITED_TEMPLATE } from "../entitlements/catalog.js"; @@ -1094,6 +1094,159 @@ describe("durable multi-step workflow engine", () => { assert.equal(run.deadlineAt.toISOString(), "2026-08-06T12:02:00.000Z"); }); + it("derives workspace affinity from the authenticated conversation and retains it through the workflow deadline", async () => { + const now = new Date("2026-08-06T12:00:00.000Z"); + const fixture = await workflowFixture({ rawConfiguration: affinityConfiguration() }); + const baseProvider = providerMatch(fixture.configuration, fixture.revisionId); + const provider = { + ...baseProvider, + workspaceAffinityKey: () => "manual-conversation-7", + } satisfies import("../triggers/index.js").TriggerProvider; + let dispatched: LaunchMachineIntent | undefined; + const { handler, engine } = createDurableWorkflowHandler({ + database: fixture.database, + entitlements: fixture.entitlements, + providers: [provider], + now: () => now, + dispatchLaunchMachineIntent: async (intent) => { + dispatched = intent; + const execution = await fixture.database.findAgentExecutionByWorkflowStepRunId( + intent.workflowStepRunId!, + ); + if (execution === undefined) throw new Error("workflow execution was not persisted"); + return { execution }; + }, + }); + + await handler(fixture.trigger("run")); + await engine.processAvailable(); + + assert.deepEqual(dispatched?.workspaceAffinity, { + key: " review-manual-conversation-7 ", + retainUntil: "2026-08-06T12:02:00.000Z", + autoArchive: true, + }); + const run = ( + await fixture.database.findTriggerRunsByProviderEventReceiptId(fixture.providerEventReceiptId) + )[0]!; + const step = (await fixture.database.listWorkflowStepRunsForTriggerRun(run.id))[0]!; + assert.deepEqual( + (await fixture.database.findAgentExecutionByWorkflowStepRunId(step.id))?.launchIntent + ?.workspaceAffinity, + dispatched?.workspaceAffinity, + ); + }); + + it("resolves conversation affinity through transitive named values", async () => { + const fixture = await workflowFixture({ + rawConfiguration: affinityConfiguration(" review-${{ values.conversation }} ", { + conversation: "${{ values.provider_conversation }}", + provider_conversation: "${{ paseo.trigger.conversation_key }}", + }), + }); + const provider = { + ...providerMatch(fixture.configuration, fixture.revisionId), + workspaceAffinityKey: () => "thread-through-values", + } satisfies import("../triggers/index.js").TriggerProvider; + let dispatched: LaunchMachineIntent | undefined; + const { handler, engine } = createDurableWorkflowHandler({ + database: fixture.database, + entitlements: fixture.entitlements, + providers: [provider], + dispatchLaunchMachineIntent: async (intent) => { + dispatched = intent; + const execution = await fixture.database.findAgentExecutionByWorkflowStepRunId( + intent.workflowStepRunId!, + ); + if (execution === undefined) throw new Error("workflow execution was not persisted"); + return { execution }; + }, + }); + + await handler(fixture.trigger("run")); + await engine.processAvailable(); + + assert.equal(dispatched?.workspaceAffinity?.key, " review-thread-through-values "); + const run = ( + await fixture.database.findTriggerRunsByProviderEventReceiptId(fixture.providerEventReceiptId) + )[0]!; + const step = (await fixture.database.listWorkflowStepRunsForTriggerRun(run.id))[0]!; + const execution = await fixture.database.findAgentExecutionByWorkflowStepRunId(step.id); + assert.ok(execution); + await fixture.database.transitionAgentExecution(execution.id, "succeeded", { + result: { status: "succeeded" }, + }); + await fixture.database.completeWorkflowStep(execution.id, "succeeded", { + status: "succeeded", + }); + await engine.processAvailable(); + assert.equal((await fixture.database.findTriggerRunById(run.id))?.status, "succeeded"); + }); + + it("fails an invalid rendered workspace affinity key without retrying or dispatching", async () => { + const fixture = await workflowFixture({ rawConfiguration: affinityConfiguration() }); + const baseProvider = providerMatch(fixture.configuration, fixture.revisionId); + const provider = { + ...baseProvider, + workspaceAffinityKey: () => "x".repeat(506), + } satisfies import("../triggers/index.js").TriggerProvider; + const dispatches: LaunchMachineIntent[] = []; + const { handler, engine } = createDurableWorkflowHandler({ + database: fixture.database, + entitlements: fixture.entitlements, + providers: [provider], + dispatchLaunchMachineIntent: async (intent) => { + dispatches.push(intent); + throw new Error("invalid workspace affinity key must not dispatch"); + }, + }); + + await handler(fixture.trigger("run")); + await engine.processAvailable(); + await engine.processAvailable(); + + const run = ( + await fixture.database.findTriggerRunsByProviderEventReceiptId(fixture.providerEventReceiptId) + )[0]!; + const step = (await fixture.database.listWorkflowStepRunsForTriggerRun(run.id))[0]!; + assert.equal(run.status, "failed"); + assert.match(run.failureReason ?? "", /workspace affinity key exceeds 512 characters/iu); + assert.equal(step.status, "failed"); + assert.deepEqual(dispatches, []); + }); + + it("does not resolve provider affinity for a literal workspace key", async () => { + const fixture = await workflowFixture({ + rawConfiguration: affinityConfiguration(" shared-release-triage "), + }); + const baseProvider = providerMatch(fixture.configuration, fixture.revisionId); + const provider = { + ...baseProvider, + workspaceAffinityKey: () => { + throw new Error("literal affinity must not query the provider"); + }, + } satisfies import("../triggers/index.js").TriggerProvider; + let dispatched: LaunchMachineIntent | undefined; + const { handler, engine } = createDurableWorkflowHandler({ + database: fixture.database, + entitlements: fixture.entitlements, + providers: [provider], + dispatchLaunchMachineIntent: async (intent) => { + dispatched = intent; + const execution = await fixture.database.findAgentExecutionByWorkflowStepRunId( + intent.workflowStepRunId!, + ); + if (execution === undefined) throw new Error("workflow execution was not persisted"); + return { execution }; + }, + }); + + await handler(fixture.trigger("run")); + await engine.processAvailable(); + + assert.equal(dispatched?.workspaceAffinity?.key, " shared-release-triage "); + }); + it("times out a live step when the whole-run deadline expires", async () => { let now = new Date("2026-08-06T12:00:00.000Z"); const fixture = await workflowFixture({ @@ -1465,6 +1618,51 @@ describe("durable multi-step workflow engine", () => { const execution = await fixture.database.findAgentExecutionByWorkflowStepRunId(steps[0]!.id); assert.equal(execution?.launchIntent?.prompt, 'Context: {"ambient":"once"}\nTrigger: run'); }); + + it("does not resolve conversation affinity when recovering a persisted pre-handoff execution", async () => { + let now = new Date("2026-08-06T12:00:00.000Z"); + const fixture = await workflowFixture({ rawConfiguration: affinityConfiguration() }); + let affinityLookups = 0; + let dispatches = 0; + const provider = { + ...providerMatch(fixture.configuration, fixture.revisionId), + workspaceAffinityKey() { + affinityLookups += 1; + if (affinityLookups > 1) throw new Error("provider registration is unavailable"); + return "conversation-once"; + }, + } satisfies import("../triggers/index.js").TriggerProvider; + const { handler, engine } = createDurableWorkflowHandler({ + database: fixture.database, + entitlements: fixture.entitlements, + providers: [provider], + now: () => now, + leaseMs: 1_000, + dispatchLaunchMachineIntent: async (intent) => { + dispatches += 1; + if (dispatches === 1) throw new Error("dispatch crashed after persistence"); + const execution = await fixture.database.findAgentExecutionByWorkflowStepRunId( + intent.workflowStepRunId!, + ); + if (execution === undefined) throw new Error("workflow execution was not persisted"); + return { execution }; + }, + }); + + await handler(fixture.trigger("run")); + await engine.processAvailable(); + now = new Date("2026-08-06T12:00:02.000Z"); + await engine.processAvailable(); + + assert.equal(affinityLookups, 1); + assert.equal(dispatches, 2); + const run = ( + await fixture.database.findTriggerRunsByProviderEventReceiptId(fixture.providerEventReceiptId) + )[0]!; + const steps = await fixture.database.listWorkflowStepRunsForTriggerRun(run.id); + const execution = await fixture.database.findAgentExecutionByWorkflowStepRunId(steps[0]!.id); + assert.equal(execution?.launchIntent?.workspaceAffinity?.key, " review-conversation-once "); + }); }); interface Fixture { @@ -1512,6 +1710,7 @@ async function workflowFixture( : { resolvedPromptPartials: options.resolvedPromptPartials }), ...(options.namedAgents === undefined ? {} : { namedAgents: options.namedAgents }), }); + const eventSource = compiled.triggers[0]?.on ?? "manual.run"; const configuration: CompiledHubConfig = { environments: compiled.environments.map((environment) => { if (environment.kind !== "daemon") return environment; @@ -1539,7 +1738,7 @@ async function workflowFixture( organizationId: "org-1", projectId: project.id, deliveryId: randomUUID(), - source: "manual.run", + source: eventSource, payload: {}, receivedAt: new Date(), }); @@ -1558,7 +1757,7 @@ async function workflowFixture( organizationId: "org-1", projectId: project.id, configurationRevisionId: revision.id, - source: "manual.run", + source: eventSource, deliveryId: receipt.event.deliveryId, payload: { input: message }, receivedAt: new Date(), @@ -1660,6 +1859,38 @@ function deadlineConfiguration(options: { idleTimeout?: string } = {}): Record>, +): Record { + return { + environments: [{ name: "runner", kind: "daemon", daemon: "runner", cwd: "/workspace" }], + triggers: [ + { + name: "affinity-route", + on: "slack.mention", + max_runtime: "2m", + filters: { from_users: ["U_ALLOWED"] }, + ...(values === undefined ? {} : { values }), + steps: [ + { + id: "review", + environment: "runner", + max_runtime: "1m", + idle_timeout: "20s", + agent: { provider: "codex" }, + prompt: [{ text: "run" }], + auto_archive: true, + workspace_affinity: { + key, + }, + }, + ], + }, + ], + }; +} + function contextOptInConfiguration(): Record { return { environments: [{ name: "runner", kind: "daemon", daemon: "runner", cwd: "/workspace" }], @@ -1893,9 +2124,12 @@ function finalValueConfiguration(): Record { } function providerMatch(configuration: CompiledHubConfig, revisionId: string) { + const eventName = configuration.triggers[0]?.on ?? "manual.run"; + assertTriggerEventName(eventName); + const providerName = eventName.slice(0, eventName.indexOf(".")); return { - name: "manual", - eventNames: ["manual.run"] as const, + name: providerName, + eventNames: [eventName], async match(external): Promise { const trigger = configuration.triggers[0]!; const input = @@ -1908,8 +2142,8 @@ function providerMatch(configuration: CompiledHubConfig, revisionId: string) { return [ { triggerName: trigger.name, - triggerContext: { provider: "manual" }, - outputContext: { provider: "manual" }, + triggerContext: { provider: providerName }, + outputContext: { provider: providerName }, configurationRevisionId: revisionId, hubConfig: configuration, conversation: null, @@ -1977,6 +2211,10 @@ function baseConfiguration(options: { unavailableValue?: boolean }): Record { return { environments: [{ name: "runner", kind: "daemon", daemon: "runner", cwd: "/workspace" }], diff --git a/src/workflows/engine.ts b/src/workflows/engine.ts index 77227ac9..90e06cd6 100644 --- a/src/workflows/engine.ts +++ b/src/workflows/engine.ts @@ -38,10 +38,13 @@ import { asTriggerContextValue, isAcceptedTriggerProviderMatch } from "../trigge import { ExpressionEvaluationError, evaluateExpression, + expressionPaths, expressionPathsInTemplate, renderExecutionTemplate, renderExpressionTemplate, + type Expression, type ExpressionContext, + type ExpressionPath, } from "./expression.js"; import type { WorktreeTarget } from "../config/index.js"; import type { Logger } from "pino"; @@ -605,6 +608,12 @@ export class DurableWorkflowEngine { if (existing?.launchIntent !== null && existing?.launchIntent !== undefined) { return { executionId: existing.id, intent: existing.launchIntent }; } + const triggerConversationKey = workspaceAffinityConversationKey( + trigger, + step, + this.options.providers ?? [], + run.triggerContext, + ); const executionId = durableExecutionId({ triggerRunId: run.id, configurationRevisionId: run.configurationRevisionId, @@ -618,7 +627,11 @@ export class DurableWorkflowEngine { trigger, step, run, - { ...context, context: materializedContext }, + { + ...context, + context: materializedContext, + ...(triggerConversationKey === undefined ? {} : { triggerConversationKey }), + }, stepRunId, deadlineAt, executionId, @@ -930,6 +943,16 @@ function buildStepIntent( throw new Error(`workflow environment ${environmentName} is unavailable`); } const agent = materializeAgent(step.agent, context); + const workspaceAffinity = + step.workspaceAffinity === undefined + ? undefined + : { + key: workspaceAffinityKey(renderExpressionTemplate(step.workspaceAffinity.key, context)), + // A matching arrival extends this deadline on the daemon. It is deliberately the + // workflow's hard deadline rather than a separate sliding inactivity timer. + retainUntil: run.deadlineAt.toISOString(), + autoArchive: step.autoArchive, + }; return { ...buildLaunchMachineIntent({ organizationId: run.organizationId, @@ -959,6 +982,7 @@ function buildStepIntent( idleTimeoutMs: step.idleTimeoutMs, ...(step.startupTimeoutMs === undefined ? {} : { startupTimeoutMs: step.startupTimeoutMs }), autoArchive: step.autoArchive, + ...(workspaceAffinity === undefined ? {} : { workspaceAffinity }), triggerContext: run.triggerContext, outputContext: run.outputContext, configurationRevisionId: run.configurationRevisionId, @@ -1018,6 +1042,16 @@ function workflowContext( }; } +function workspaceAffinityKey(value: string): string { + if (value.trim().length === 0) { + throw new ExpressionEvaluationError("workspace affinity key resolved to an empty value"); + } + if (value.length > 512) { + throw new ExpressionEvaluationError("workspace affinity key exceeds 512 characters"); + } + return value; +} + function stepUsesTriggerContext( step: CompiledProjectConfiguration["triggers"][number]["steps"][number], ): boolean { @@ -1039,15 +1073,73 @@ function providerForTriggerContext( return providers.find((provider) => provider.name === triggerContext.provider); } +function workspaceAffinityConversationKey( + trigger: CompiledProjectConfiguration["triggers"][number], + step: CompiledProjectConfiguration["triggers"][number]["steps"][number], + providers: readonly TriggerProvider[], + triggerContext: unknown, +): string | undefined { + if (!stepUsesTriggerConversationKey(trigger, step)) return undefined; + return providerForTriggerContext(providers, triggerContext)?.workspaceAffinityKey?.( + triggerContext, + ); +} + +function stepUsesTriggerConversationKey( + trigger: CompiledProjectConfiguration["triggers"][number], + step: CompiledProjectConfiguration["triggers"][number]["steps"][number], +): boolean { + if (step.workspaceAffinity === undefined) return false; + return pathsUseTriggerConversationKey( + expressionPathsInTemplate(step.workspaceAffinity.key), + trigger.values, + ); +} + +function expressionUsesTriggerConversationKey( + expression: Expression, + values: Readonly>, +): boolean { + return pathsUseTriggerConversationKey(expressionPaths(expression), values); +} + +function pathsUseTriggerConversationKey( + paths: readonly ExpressionPath[], + values: Readonly>, +): boolean { + const pending = [...paths]; + const visitedValues = new Set(); + while (pending.length > 0) { + const path = pending.pop(); + if (path === undefined) continue; + if ( + path.namespace === "paseo" && + Array.isArray(path.path) && + path.path[0] === "trigger" && + path.path[1] === "conversation_key" + ) { + return true; + } + if (path.namespace !== "values" || visitedValues.has(path.name)) continue; + visitedValues.add(path.name); + const expression = values[path.name]; + if (expression !== undefined) pending.push(...expressionPaths(expression)); + } + return false; +} + function composeValues( values: Readonly>, context: ExpressionContext, ): Readonly> { return Object.fromEntries( - Object.entries(values).map(([name, expression]) => [ - name, - evaluateExpression(expression, context), - ]), + Object.entries(values) + .filter( + ([, expression]) => + context.triggerConversationKey !== undefined || + !expressionUsesTriggerConversationKey(expression, values), + ) + .map(([name, expression]) => [name, evaluateExpression(expression, context)]), ); } diff --git a/src/workflows/expression.test.ts b/src/workflows/expression.test.ts index 88dbf120..264ad9cb 100644 --- a/src/workflows/expression.test.ts +++ b/src/workflows/expression.test.ts @@ -36,4 +36,35 @@ describe("workflow expression context", () => { "trigger-64ae56ff-281c-4c5f-bf5c-d572f125c702", ); }); + + it("renders the provider-authenticated conversation key only when supplied", () => { + const context = { + prompt: "the triggering body", + context: {}, + inputs: {}, + steps: {}, + values: {}, + triggerConversationKey: "slack:thread:1700000000.000001", + }; + + assert.equal( + renderExpressionTemplate("thread-${{ paseo.trigger.conversation_key }}", context), + "thread-slack:thread:1700000000.000001", + ); + assert.deepEqual(parseExpression("${{ paseo.trigger.conversation_key }}"), { + kind: "path", + value: { namespace: "paseo", path: ["trigger", "conversation_key"] }, + }); + assert.throws( + () => + renderExpressionTemplate("${{ paseo.trigger.conversation_key }}", { + prompt: context.prompt, + context: context.context, + inputs: context.inputs, + steps: context.steps, + values: context.values, + }), + /trigger conversation key is unavailable/iu, + ); + }); }); diff --git a/src/workflows/expression.ts b/src/workflows/expression.ts index adeaea49..cc913603 100644 --- a/src/workflows/expression.ts +++ b/src/workflows/expression.ts @@ -3,7 +3,12 @@ import type { JsonPrimitive, JsonValue } from "../config/compiler.js"; export type ExpressionPath = | { namespace: "paseo"; - path: "prompt" | "context" | ["inputs", string] | ["execution", "id"]; + path: + | "prompt" + | "context" + | ["inputs", string] + | ["execution", "id"] + | ["trigger", "conversation_key"]; } | { namespace: "steps"; stepId: string; path: readonly string[] } | { namespace: "values"; name: string }; @@ -25,6 +30,7 @@ export interface ExpressionContext { inputs: Readonly>; steps: Readonly>; values: Readonly>; + triggerConversationKey?: string; executionId?: string; } @@ -186,6 +192,12 @@ function parsePaseoPath(parts: readonly string[]): Expression { if (parts[1] === "execution" && parts[2] === "id" && parts.length === 3) { return { kind: "path", value: { namespace: "paseo", path: ["execution", "id"] } }; } + if (parts[1] === "trigger" && parts[2] === "conversation_key" && parts.length === 3) { + return { + kind: "path", + value: { namespace: "paseo", path: ["trigger", "conversation_key"] }, + }; + } throw new ExpressionSyntaxError(`unsupported path ${parts.join(".")}`); } @@ -386,6 +398,12 @@ function readPath(path: ExpressionPath, context: ExpressionContext): JsonValue { } return context.executionId; } + if (path.path[0] === "trigger") { + if (context.triggerConversationKey === undefined) { + throw new ExpressionEvaluationError("trigger conversation key is unavailable"); + } + return context.triggerConversationKey; + } return context.inputs[path.path[1]] ?? null; } if (path.namespace === "values") {