Phase 5 — Async Job System (Inngest)

Goal

The agent can hand off long-running work and learn when it’s done via push, without the MCP server blocking. After this phase, the system supports the full async loop: create_job → Inngest worker → notifications/job_status pushed to the originating MCP session.

Scope / Deliverables

• Inngest keys (INNGEST_EVENT_KEY + INNGEST_SIGNING_KEY) in env, validated by config.ts
• POST /api/inngest mounted with the SDK’s serve() helper
• Migration: jobs table (full schema from data-and-contracts.md)
• jobs/events.ts — Zod schemas for job.requested, job.completed, job.failed, job.cancelled
• jobs/inngest.ts — typed Inngest client (event sender)
• First worker function: image.generate — concurrency key = userId, single step.run placeholder that returns a stub URL (real image-gen provider deferred)
• Tool: create_job — writes the jobs row (status='queued'), captures mcp_session_id, emits job.requested
• Tool: get_job_status — reads from jobs row
• Tool: cancel_job — sets status='cancelled' in DB; worker checks status between steps and exits early
• Worker function on success → emits job.completed and updates jobs row to succeeded with result
• Worker function on terminal failure → emits job.failed and updates jobs row to failed with error
• System function notify.job_status triggered by job.completed|failed|cancelled → looks up mcp_session_id → pushes JSON-RPC notifications/job_status to that session (drops silently if not connected)
• mcp/sessions.ts — connect/disconnect handlers maintain the in-process session map

See data-and-contracts.md for the MCP session & push model and the full Inngest event schemas.

Automated Unit Testing

• events.schema.test.ts — for each event (job.requested, job.completed, job.failed, job.cancelled):
  • valid payload parses cleanly
  • missing required field → Zod error with field path
  • extra unknown fields handled per the chosen Zod stance (passthrough vs strict — document)
  • job.failed.error requires both code and message
• tools/create_job.test.ts —
  • writes a jobs row with status='queued', generated UUID, captured mcp_session_id, and the typed payload
  • emits exactly one job.requested event via the (mocked) Inngest client, with matching jobId, userId, mcpSessionId, type, payload
  • unknown type → INVALID_INPUT (or accepted as opaque string — document the policy)
  • Zod rejects payloads not matching the per-type schema (when registered)
• tools/get_job_status.test.ts —
  • reads the row and shapes it as {jobId, type, status, result?, error?, createdAt, updatedAt}
  • unknown jobId → NOT_FOUND
• tools/cancel_job.test.ts —
  • while status='queued' → flips to cancelled, returns {jobId, status: 'cancelled'}
  • while status='running' → flips to cancelled
  • while status='succeeded'|'failed'|'cancelled' → rejects with INVALID_INPUT (or stable error), no DB write
• sessions.test.ts —
  • register(sessionId, pushFn) populates the map
  • unregister(sessionId) clears it
  • pushTo(sessionId, msg) invokes the pushFn for an active session and is a silent no-op for an absent one
• notify.job_status.test.ts —
  • given a job.completed event, resolves the row’s mcp_session_id and invokes the right pushFn with { method: 'notifications/job_status', params: { jobId, status: 'succeeded', result } }
  • given a job.failed event, push includes { status: 'failed', error: { code, message } }
  • if the session is absent from the map, no error is thrown (drop silently)
• jobs/imageGenerate.test.ts (Inngest function, run via inngest.test harness or pure unit on the wrapped business logic):
  • happy path → job.completed emitted, jobs row updated to succeeded with result
  • mid-run DB shows status='cancelled' → function exits early at the next step.run checkpoint without emitting job.completed
  • thrown NonRetriableError from a step → job.failed emitted, jobs.error populated

Integration scope (deferred): full Inngest local dev round-trip (npx inngest-cli dev + a real event) is exercised in the integration suite.

Definition of Done

• Agent calls create_job(type='image.generate', payload={prompt}) → gets jobId immediately.
• Inngest dashboard shows the function ran with the right concurrency key.
• Agent receives a notifications/job_status push on completion, within ~1s of the function finishing.
• Disconnecting and reconnecting the agent: get_job_status(jobId) reflects the final state.
• Two create_job calls for the same userId run sequentially, not in parallel.