feat(decopilot): per-thread DBOS gate for user messages

[viktormarinho]

Summary

Adds a per-thread DBOS gate so agent runs on the same thread execute one at a time, while runs on different threads progress in parallel. Both user messages and automation fires now funnel through the same gate.

POST /:org/decopilot/threads/:threadId/messages enqueues onto a partitioned DBOS workflow (threadGateWorkflow, partition key = threadId, concurrency = 1) and returns 202 { taskId } in milliseconds. A second POST while a run is in flight queues behind it and dispatches only after the active run completes. Streaming continues through GET /:org/decopilot/attach/:threadId.

Changes

apps/mesh/src/dispatch-queue/ (new)

• threadGateWorkflow — single DBOS workflow that owns the per-thread serialization. Body is two steps:
  1. trackMessageStarted — emits chat_message_started PostHog event (skipped for automation sources). Wrapped in a step so idempotent retries that collapse onto an existing workflow ID don't double-count.
  2. dispatchRunAndWait — invokes the configured dispatch fn; constructs its own AbortController (since AbortSignal isn't serializable across workflow boundaries). If the step throws, trackMessageFailed emits with error_category: "setup" to balance the started event.
• enqueueThreadRun(ctx, { workflowID? }) — fire-and-forget. Used by the user-message route.
• awaitThreadRun(ctx, { workflowID? }) — enqueue + getResult(). Used by automation fires that need to block on the inner run's outcome.
• Module-level setThreadGateRuntime() registry wires dispatchRunFn, meshContextFactory, and dispatch deps before DBOS.launch().
• THREAD_GATE_QUEUE + THREAD_GATE_PARTITION_CONCURRENCY constants; queue registered in createApp.
• Abort timer is opt-in: automations pass an explicit 5 min cap so a runaway cron can't pin a thread slot; user messages leave it unset because tool-using loops (Claude Code, deep research) routinely outlast any fixed cap and weren't bounded by the legacy fire-and-forget path.

apps/mesh/src/api/routes/decopilot/routes.ts

• POST /messages replaces the inline dispatchAndTrack call with enqueueThreadRun. Returns 202 { taskId }.
• Idempotency: prefers X-Idempotency-Key header, falls back to the last message's id. Combined as workflowID = thread-run:<threadId>:<key>. Without either, retries get a fresh workflow ID (at-least-once).
• dispatchAndTrack deleted; PostHog chat_message_started is now emitted from the workflow step.
• Orphan-resume in /attach continues to call dispatchRun directly — going through the gate would deadlock against itself.

apps/mesh/src/automations/dbos-workflow.ts

• fireAutomationWorkflowFn hands off to the thread gate via awaitThreadRun instead of calling dispatchRunFn directly. Existing per-automation (concurrency=3) and global (concurrency=5) gates remain layered above; the per-thread gate adds the third layer.
• Split the dispatch into two phases so DBOS rules are respected:
  • buildDispatchRequest step — membership pre-check + buildStreamRequest, journaled so replays reuse the same crypto.randomUUID() message ids.
  • awaitThreadRun from the workflow body — DBOS forbids invoking a workflow from inside a step, so this lives in the workflow context. Failures are caught at the workflow level to preserve the FireAutomationOutcome {taskId, error} shape.
• markRunFailed extracted into its own step so the side effect is recorded on the journal.
• AutomationRuntime loses dispatchRunFn and deps (now owned by the thread-gate runtime).

apps/mesh/src/api/app.ts

• Imports setThreadGateRuntime, THREAD_GATE_QUEUE, THREAD_GATE_PARTITION_CONCURRENCY from @/dispatch-queue.
• Calls setThreadGateRuntime({ dispatchRunFn: dispatchRunAndWait, meshContextFactory, deps }) before DBOS.launch().
• Registers the thread-gate queue with partitionQueue: true, concurrency: 1.

apps/mesh/src/automations/fire.ts

• Removes the now-unused DispatchRunFn type and DispatchRunInput/DispatchRunDeps imports.

Telemetry semantics

• chat_message_started — fires from inside the workflow body (post queue-admission), once per workflow. Idempotent retries collapse onto the same workflowID and do not double-count.
• chat_message_failed — emitted from one of three places, never two:
  • Setup errors thrown by prepareRun before streamText starts → workflow's trackMessageFailed step with error_category: "setup".
  • Mid-stream errors → createUIMessageStream.onError inside dispatchRun with classifyStreamError(error) category.
  • Automation fires are excluded from message-send analytics.

Idempotency contract

• Header: X-Idempotency-Key: <stable-key> (recommended for retrying clients).
• Fallback: the last (user) message's id. Most chat clients already send a UUID per message.
• Without either, retries are at-least-once. Clients that need exactly-once must send one of the two.
• Format: thread-run:<threadId>:<key>. A redelivered POST gets 202 and getResult()-equivalent semantics inside DBOS.

What's NOT in this PR

• Inbox UI + GET/DELETE /threads/:id/queue endpoints — follow-up.
• Orphan-resume continues to bypass the gate (recovery path).

Bug fixes included on the branch

• DBOS step/workflow invariant (commit ff8914bb0): the initial automation routing crashed every fire with Invalid call to a 'workflow' function from within a 'step' or 'transaction' because awaitThreadRun was called from inside DBOS.runStep. Fixed by moving awaitThreadRun to the workflow body and splitting the prep work into a journaled step.

Test plan

• bun run --cwd=apps/mesh check clean (TypeScript).
• bun run fmt clean.
• thread-gate-workflow.test.ts covers queue plumbing (constant exports + setThreadGateRuntime runtime shape).
• Manual: send a user message; verify 202 { taskId } and stream attaches via /attach.
• Manual: send two user messages back-to-back on the same thread; verify they serialize.
• Manual: trigger an automation fire; verify it dispatches through the thread gate without the DBOS step/workflow error.
• Manual: retry a POST with the same X-Idempotency-Key; verify no duplicate run / no duplicate chat_message_started.

🤖 Generated with Claude Code

[github-actions]

🧪 Benchmark

Should we run the Virtual MCP strategy benchmark for this PR?

React with 👍 to run the benchmark.

Reaction	Action
👍	Run quick benchmark (10 & 128 tools)

Benchmark will run on the next push after you react.

[github-actions]

Release Options

Suggested: Minor (2.329.0) — based on feat: prefix

React with an emoji to override the release type:

Reaction	Type	Next Version
👍	Prerelease	2.328.1-alpha.1
🎉	Patch	2.328.1
❤️	Minor	2.329.0
🚀	Major	3.0.0

Current version: 2.328.0

Note: If multiple reactions exist, the smallest bump wins. If no reactions, the suggested bump is used (default: patch).

[cubic-dev-ai]

2 issues found across 5 files

Prompt for AI agents (unresolved issues)

Check if these issues are valid — if so, understand the root cause of each and fix them. If appropriate, use sub-agents to investigate and fix each issue separately.


<file name="apps/mesh/src/dispatch-queue/thread-gate-workflow.ts">

<violation number="1" location="apps/mesh/src/dispatch-queue/thread-gate-workflow.ts:127">
P1: Do not swallow dispatch exceptions in the workflow step; rethrow so failed runs are recorded as workflow failures (and can follow normal retry/failure handling) instead of succeeding with an ignored `{ error }` payload.</violation>
</file>

<file name="apps/mesh/src/api/routes/decopilot/routes.ts">

<violation number="1" location="apps/mesh/src/api/routes/decopilot/routes.ts:328">
P2: `chat_message_started` is emitted unconditionally after enqueue, so idempotent POST retries can be double-counted even when they collapse to an existing workflow.</violation>
</file>

_{Reply with feedback, questions, or to request a fix. Tag @cubic-dev-ai to re-run a review.
Re-trigger cubic}

[cubic-dev-ai]

1 issue found across 2 files (changes from recent commits).

Prompt for AI agents (unresolved issues)

Check if these issues are valid — if so, understand the root cause of each and fix them. If appropriate, use sub-agents to investigate and fix each issue separately.


<file name="apps/mesh/src/api/routes/decopilot/routes.ts">

<violation number="1" location="apps/mesh/src/api/routes/decopilot/routes.ts:303">
P2: Empty `X-Idempotency-Key` values block fallback to message id, which can silently disable idempotency on retries.</violation>
</file>

_{Tip: Review your code locally with the cubic CLI to iterate faster.
Re-trigger cubic}