system-bus

name: system-bus displayName: System Bus Worker description: Develop, deploy, and debug the system-bus worker — joelclaw's 110+ Inngest durable function engine, webhook gateway, and observability pipeline. Triggers on 'add a function', 'new inngest function', 'system-bus', 'worker', 'add a webhook', 'deploy worker', 'restart worker', 'function failed', 'worker not working', 'register functions', or any task involving Inngest function development, webhook providers, or worker operations. version: 0.1.5 author: joel tags: - inngest - worker - infrastructure - core

System Bus Worker

The system-bus worker (@joelclaw/system-bus) is joelclaw's event-driven backbone — 110+ Inngest durable functions, webhook ingestion, and observability. It runs as a Hono HTTP server registered with the self-hosted Inngest instance.

Architecture

packages/system-bus/
├── src/
│   ├── serve.ts                          # Hono server, Inngest registration, health endpoint
│   ├── inngest/
│   │   ├── client.ts                     # Inngest client + event type definitions
│   │   ├── middleware/                    # Gateway injection, dependency injection
│   │   └── functions/
│   │       ├── index.ts                  # Combined exports
│   │       ├── index.host.ts             # Functions for host-role worker (local Mac)
│   │       ├── index.cluster.ts          # Functions for cluster-role worker (k8s)
│   │       └── <function-name>.ts        # Individual functions
│   ├── lib/                              # Shared utilities
│   │   ├── inference.ts                  # LLM calls via pi (CANONICAL — always use this)
│   │   ├── redis.ts                      # Redis client helper
│   │   ├── typesense.ts                  # Typesense client
│   │   ├── convex-content-sync.ts        # Convex upsert for content pipeline
│   │   ├── langfuse.ts                   # Langfuse tracing
│   │   └── ...
│   ├── observability/
│   │   └── emit.ts                       # OTEL event emission
│   ├── webhooks/
│   │   ├── server.ts                     # Webhook router (mounted at /webhooks)
│   │   ├── types.ts                      # Provider interface
│   │   └── providers/                    # Per-service webhook handlers
│   │       ├── front.ts
│   │       ├── github.ts
│   │       ├── vercel.ts
│   │       ├── todoist.ts
│   │       ├── mux.ts
│   │       └── joelclaw.ts
│   └── memory/                           # Memory pipeline components
├── scripts/
│   └── sync-content-to-convex.ts         # Manual full Convex sync
└── package.json

Worker Roles

Two deployment modes controlled by WORKER_ROLE env var:

Functions are split between index.host.ts and index.cluster.ts. The combined index.ts exports everything for tooling/tests.

Deployment Model

• Source of truth: ~/Code/joelhooks/joelclaw/packages/system-bus/
• Running host worker: worker-supervisor process running bun run src/serve.ts from ~/Code/joelhooks/joelclaw/packages/system-bus/
  • verify with lsof -iTCP:3111 -sTCP:LISTEN -n -P and lsof -p <pid> | awk '$4=="cwd"{print}'
  • legacy clone ~/Code/system-bus-worker/ may still exist, but it is not the active host worker when port 3111's cwd points at the monorepo
• Cluster runtime: system-bus-worker Deployment in the Talos/Colima k8s cluster for cluster-role workloads
• Cluster deploy path: ~/Code/joelhooks/joelclaw/k8s/publish-system-bus-worker.sh

Host function rollout reality

Host worker registration supports an explicit INNGEST_SERVE_HOST override in ~/.config/system-bus.env. Set INNGEST_SERVE_HOST=connect to suppress SDK callback URL advertising when the self-hosted Inngest pod cannot route to the host network. Do not leave it pointing at a stale Tailscale or Docker-host address unless you have proven the Inngest pod can open that address from inside k8s; otherwise every host function fails with Unable to reach SDK URL.

After changing packages/system-bus/src/inngest/functions/* that run on the host worker:

1. commit + push the monorepo change to origin
2. confirm the live worker cwd: pid=$(lsof -tiTCP:3111 -sTCP:LISTEN); lsof -p "$pid" | awk '$4=="cwd"{print}'
3. if cwd is ~/Code/joelhooks/joelclaw/packages/system-bus, kill the Bun worker PID and let worker-supervisor respawn it
4. if cwd is the legacy ~/Code/system-bus-worker, sync that clone first (git fetch origin && git reset --hard origin/main) and then restart the process
5. verify curl http://127.0.0.1:3111/ shows functions and joelclaw functions returns >0

The stale failure mode: a host worker can keep running old source for days. In that state, OTEL may show behavior that current monorepo code has already fixed. Always verify the live port-3111 process cwd and start time before debugging source that "should" already be deployed.

Queue pilot flags are evaluated inside the live worker process, not your shell. If a host-worker emitter like discovery-capture or /webhooks/github should switch to queue mode, put the flag in ~/.config/system-bus.env, then kickstart the worker and PUT-sync /api/inngest. Ad-hoc shell env only affects CLI-local emitters.

Queue triage flags follow the same rule. Current bounded admission contract:

• QUEUE_TRIAGE_MODE=off|shadow|enforce sets the base triage mode.
• QUEUE_TRIAGE_FAMILIES=discovery,content,subscriptions,github (or exact event names) chooses which queue families participate at all.
• QUEUE_TRIAGE_ENFORCE_FAMILIES=discovery,github is the narrow Story 4 override that upgrades only discovery/noted and github/workflow_run.completed into enforce.
• Any non-eligible family is clamped back to shadow even if someone sets global QUEUE_TRIAGE_MODE=enforce.
• Handler routing always stays registry-derived; triage may only shape bounded admission fields.

content/updated is the odd one out: its ingress comes from the launchd watcher com.joel.content-sync-watcher, not from a worker-local function. The canonical watcher source now belongs in infra/launchd/com.joel.content-sync-watcher.plist plus scripts/content-sync-watcher.sh, and the script reads ~/.config/system-bus.env on each trigger so QUEUE_PILOTS=content can switch between joelclaw queue emit and legacy joelclaw send without hand-editing the live plist.

For Story 5 soak work, start from joelclaw jobs status for the first operator glance, then drop into joelclaw queue stats before spelunking raw OTEL or Redis. jobs status is the transitional runtime surface that rolls queue / Restate / Dkron / Inngest into one JSON snapshot without forcing the operator to jump between commands just to learn whether the substrate is healthy enough to take work. queue stats remains the queue-specific summary for Restate drainer health and queue triage behavior: it rolls up recent queue.dispatch.started|completed|failed telemetry plus the queue.triage.* lifecycle into live depth, terminal success/failure counts, waitTimeMs percentiles, dispatch-duration percentiles, fallback reasons, disagreement counts, applied-vs-suggested deltas, route mismatches, family rollups, and recent mismatch/fallback samples. Use joelclaw queue stats --since <iso|ms> when you need to anchor the sample to a known-clean point such as a supervised queue.drainer.started after rollout. Honest gotcha from the live Story 5 cleanup follow-through: global depth can lie because of unrelated historical backlog, so judge the supervised sample first with the anchored triage/dispatch window plus joelclaw queue inspect <stream-id> / joelclaw queue list --limit <n> on the fresh sample IDs. If old residue survives a supervised com.joel.restate-worker restart, clear it with a bounded @joelclaw/queue ack() pass only after confirming zero pending leases and an age filter on the orphaned stream IDs. If that command is broken or misleading, fix it before widening queue cutovers.

For ADR-0217 Phase 3 Story 2-4, the operator surfaces are joelclaw queue observe, joelclaw queue pause, joelclaw queue resume, and joelclaw queue control status. queue observe still answers “what would Sonnet do right now?” in dry-run, but its snapshot.control.activePauses plus top-level control block now reflect the shipped deterministic control plane: active pauses, expirations, and recent queue.control.* OTEL. It short-circuits to a deterministic noop when the backlog is fully explained by fresh active manual pauses and no recent failures suggest downstream trouble, and it now also short-circuits to deterministic resume_family when queued work is entirely held behind a settled observer pause with no fresh drainer/triage failures. That avoids wasting a 60s Sonnet call on an obvious operator hold state and stops the observer from mistaking its own pause for permanent downstream failure. queue pause / queue resume are the bounded manual apply path before any automatic observer mutation. queue control status is the direct operator truth source for active manual controls and recent queue.control.applied|expired|rejected events.

ADR-0217 Phase 3 Story 4 now has a live host-worker runtime in packages/system-bus/src/inngest/functions/queue-observer.ts. Durable cadence belongs in Inngest, not the gateway daemon: the cron controller stays on queue/observer, while manual queue/observer.requested probes now run through a separate queue/observer-requested function so operator requests do not sit behind the cron pass. Runtime flags live in ~/.config/system-bus.env and require the usual host-worker restart + PUT /api/inngest:

• QUEUE_OBSERVER_MODE=off|dry-run|enforce
• QUEUE_OBSERVER_FAMILIES=discovery,content,subscriptions,github
• QUEUE_OBSERVER_AUTO_FAMILIES=content
• QUEUE_OBSERVER_INTERVAL_SECONDS (currently clamped to a 60s minimum on the durable cron path)

Both paths build the same bounded snapshot and use the shared Sonnet observer contract, but only the cron controller may auto-apply queue actions; manual probes are read-only even when the configured mode is enforce. The shared observer also short-circuits deterministic noops for both truly empty queues and empty queues that only still have active pauses hanging around, so the cron path does not waste a full model call on obvious nothing-to-do snapshots. Idle empty snapshots with no recent drainer failures or triage trouble now report downstreamState=healthy instead of inheriting a noisy degraded label from stale throughput/latency history. Settled observer-held backlogs now get the same treatment: once a queued family is already behind an observer pause for at least one cadence and no fresh drainer/triage failures exist, the shared observer deterministically emits resume_family instead of reading its own hold as proof that downstream is still down. The manual probe function also uses singleton-skip semantics so repeated operator requests do not pile up stale queued runs. Live canaries also exposed a prompt-contract gotcha: Sonnet was happy to emit escalate.reason, which blew up the strict schema. The shared parser now normalizes that legacy shape to { severity: "warn", message }, and the prompt explicitly tells Sonnet to prefer pause_family/resume_family over batch_family for the content/updated pilot when downstream is crook.

Current operator truth after the latest live canaries: dry-run is earned and the hardened enforce path has now completed one full automatic observer cycle on content/updated. The first supervised canary anchored at since=1772981290859 booted Restate out long enough to build a 30-item content backlog and auto-applied pause_family on snapshot cca656f7-a9ce-4ca2-9f6d-0ed332f56a4d. The patched follow-up canary anchored at since=1772985057594 then paused on snapshot 1cb24e7b-f0cd-4e0c-ae5d-27cb4934b49a, auto-resumed on snapshot 151aa03a-fced-41f0-9a54-2f3d1a70856d / run 01KK72HD0EMT3T34K8QP3SMEW9, and drained the held content item back to queue depth 0. The steady-state worker still belongs in QUEUE_OBSERVER_MODE=dry-run; enforce remains a supervised drill until more soak windows say otherwise. Dogfood follow-through also exposed that check/o11y-triage is a long-running current-state scan, not an irreplaceable per-event handler, so it now needs singleton-skip semantics to avoid piling duplicate queued runs behind one active pass.

Hard-won gotcha from the Story 3 live proof: queue operator commands must resolve Redis from the canonical CLI config (~/.config/system-bus.env → REDIS_URL) before ambient shell env. The first proof looked wrong because the shell had an unrelated Upstash REDIS_URL, so queue pause wrote control state to the wrong Redis while queue emit still hit the localhost worker/drainer queue. If the CLI and worker disagree about Redis, fix that first or your proof is bullshit.

Adding a New Inngest Function

1. Create packages/system-bus/src/inngest/functions/<name>.ts
2. Import inngest from ../client
3. Define the function:

import { inngest } from "../client";

export const myFunction = inngest.createFunction(
  {
    id: "system/my-function",
    // NEVER set retries: 0 — let Inngest defaults handle retries
    concurrency: { limit: 1, key: '"my-function"' },
  },
  { event: "my/event.name" },
  async ({ event, step, ...rest }) => {
    const gateway = (rest as any).gateway as
      import("../middleware/gateway").GatewayContext | undefined;

    const result = await step.run("do-work", async () => {
      // your logic here
      return { done: true };
    });

    return result;
  }
);

4. Export from index.host.ts or index.cluster.ts (depending on role)
5. Add the export to index.ts as well
6. Add the event type to client.ts if it's a new event
7. TypeScript check: bunx tsc --noEmit -p packages/system-bus/tsconfig.json
8. Deploy (see below)

Event Naming Convention

Events describe what happened, not commands: agent/memory.observed not agent/memory.write.

LLM Inference

ALWAYS use the shared utility:

import { infer } from "../../lib/inference";

const { text } = await infer("Your prompt", {
  task: "classification",
  model: "anthropic/claude-haiku",
  system: "System prompt here",
  component: "my-function",
  action: "my-function.classify",
  noTools: true,
  print: true,
});

This shells to pi -p --no-session --no-extensions. Zero config, zero API cost. NEVER use OpenRouter, read auth.json, or use paid API keys directly.

If the function is doing long-form editorial or other large-context LLM work, set an explicit timeout on infer() instead of inheriting the shared 120s default. Current earned example: content/review.submitted rewrite/retry/verify runs use a 300s budget because long posts can blow past 120s after all preflight bookkeeping already succeeded.

Gateway Context

All functions receive gateway context via middleware (ADR-0144). Use it for notifications:

const gateway = (rest as any).gateway as
  import("../middleware/gateway").GatewayContext | undefined;

await gateway?.notify("event.name", { details });
await gateway?.alert("Something broke", { error: String(err) });
await gateway?.progress("Step 3/5 complete");

Hard Rules

• VIP email history now has a dedicated backfill path: vip/email-threads.backfill (host worker) walks Front contact history for configured VIP sender emails and hydrates Typesense email_threads before narrative briefs rely on prior thread arcs. Default VIP sender aliases now include Alex Hillman's alex@indyhall.org address so the backfill can resolve the canonical Front contact.
• vip/email-received operator output is narrative-first now: expect 3-5 sentence prose plus one calibrated urgency line (🔴🟠🟡🟢✅), not the old Thread / Your last reply / Needs your attention block format.
• NEVER set retries: 0 — Inngest defaults handle retries. This has caused multiple production failures.
• Events silently dropped if functions not registered. Verify joelclaw functions returns >0 before sending events. joelclaw refresh forces re-registration.
• Inngest server function registry goes stale on worker restart. Always curl -X PUT http://127.0.0.1:3111/api/inngest after restart. If PUT returns Empty reply from server, check ~/.local/log/system-bus-worker.err for Bun request timeouts — the host worker must keep idleTimeout high enough for slow self-hosted Inngest registration under backlog pressure.
• Don't edit monorepo while a loop is running. git add -A scoops up unrelated changes.
• Step names must be unique within a function — Inngest uses them for memoization.
• step.invoke over fan-out events for rate-limited APIs — fan-out starts all near-simultaneously even with throttle.
• Silent failure anti-pattern: Functions that shell to CLIs must detect and propagate subprocess failures.
• ADR content sync must degrade on frontmatter parse failures. upsertAdr falls back to body-only parsing (empty frontmatter + stripped frontmatter block) and logs a warning, instead of dropping the ADR from Convex.
• Non-authoritative side effects must degrade, not crash the workflow. Example: memory/proposal-triage keeps triage authoritative, retries review-task creation across primary/fallback Todoist projects (MEMORY_REVIEW_TODOIST_PROJECT → MEMORY_REVIEW_TODOIST_FALLBACK_PROJECT), and only records degraded state if both fail.
• Never call joelclaw CLI with Bun.spawnSync from inside a running Inngest function. joelclaw inngest status probes the worker endpoint; sync subprocesses can deadlock the worker event loop. Use async subprocess execution (Bun.spawn/Bun.$) with explicit timeouts, or direct internal health probes.
• Keep Typesense URL queries bounded. Typesense rejects search query strings over 4000 bytes. Functions that search with user/email/thread text must normalize whitespace and cap q before calling typesense.search; otherwise one pasted log/email can fail retries and make the worker-supervisor restart a healthy worker. conversation-thread-enrich caps vault recall queries at 1200 chars.
• Background agent runs must be non-blocking. system/agent-dispatch cannot use execSync/other blocking subprocess APIs for long codex or claude runs on the host worker; blocking the Bun event loop causes Talon/worker-supervisor health checks to fail, the worker to restart, /internal/agent-await to drop, and Inngest runs to go stale.
• Pi is now the preferred Restate PRD story executor. system/agent-dispatch must honor the requested cwd when it calls infer(), should enable pi tools when file work is requested (readFiles or path-heavy prompts), and should use the dedicated roster agent story-executor for Restate PRD stories so they run under the tight execution prompt instead of the generic background-agent system prompt. The host bridge must also write a running inbox snapshot before long agent execution starts and dedupe /internal/agent-dispatch by requestId; otherwise multi-minute Restate retries spawn duplicate story agents and operators get a useless forever-pending state.
• Execution mode: host vs sandbox (ADR-0217 Story 4/next batch). system/agent-dispatch accepts executionMode: "host" | "sandbox" (default: "host"). Host mode uses the existing shared-checkout path. Sandbox mode now has a concrete backend split: sandboxBackend: "local" | "k8s" (default local). The local backend is the proved live path on the host worker: it materializes a clean temp checkout at baseSha, runs the requested agent inside that isolated repo, exports patch/touched-file artifacts, and then tears the sandbox down without dirtying the operator checkout. Gate A (non-coding vertical slice) is proven via packages/agent-execution/__tests__/gate-a-smoke.test.ts. Gate B (minimal coding sandbox) is proven via packages/agent-execution/__tests__/gate-b-smoke.test.ts. The k8s backend is now code-landed and opt-in: @joelclaw/agent-execution owns Job spec generation plus Job launch/status/log helpers, job-runner.ts prints SandboxExecutionResult log markers and POSTs terminal results to /internal/agent-result, and InboxResult now preserves sandboxBackend plus optional Job metadata. Current honest limit: pi remains the local-backend story executor for now; the k8s runner is for runner-installed CLIs until host-routed pi-in-pod execution is designed. Deterministic sandbox requests should carry workflowId, storyId, baseSha, repoUrl, and branch; trigger-prd.ts now has explicit tool/backend knobs (PRD_EXEC_TOOL, PRD_EXECUTION_MODE, PRD_SANDBOX_BACKEND).
• Terminal state guarantees (ADR-0217 Story 5). system/agent-dispatch ensures every execution lands in a terminal state (completed|failed|cancelled). Duplicate requests with the same requestId are deduped at function entry — if a terminal result already exists, it returns that result without spawning new work. Cancellation via system/agent.cancelled kills the active subprocess (tracked in activeProcesses map by requestId) and writes a cancelled inbox snapshot via the onFailure handler.
• Log surfacing (ADR-0217 Story 5). All terminal results include stdout/stderr output (truncated to 10KB each) in the logs field. This is captured from subprocess execution and attached to the inbox result for post-mortem debugging. The logs are also emitted via OTEL events for searchability.
• Do not capture tool-enabled pi attempts by waiting on pipe EOF. In src/lib/inference.ts, background pi runs with tools can spawn descendants that inherit stdout/stderr, leaving new Response(proc.stdout).text() or similar pipe readers hanging after the real pi child exits. Redirect stdout/stderr to temp files (or another exit-driven sink), wait for proc.exited, then read the captured output so system/agent-dispatch can always write a terminal inbox snapshot.
• Apply the same exit-driven capture rule inside system/agent-dispatch command execution. Codex/Claude/bash subprocesses and local sandbox infra commands can also leave descendants holding stdout/stderr open after the parent exits. If agent-dispatch waits on pipe EOF there, terminal inbox writeback stalls and sandbox runs lie in running even though the real work already finished or failed.
• Use tool: "canary" for deterministic live verification of the dispatch substrate itself. This is the non-LLM proof lane for system/agent-dispatch: fixed scenarios like sleep-timeout and orphan-stderr exercise the same subprocess capture + terminal inbox/registry path without depending on model behavior. Canonical timeout proof script: bun scripts/verify-agent-dispatch-timeout.ts. Canonical operator surface: joelclaw status --agent-dispatch-canary, and the default status envelope now exposes the latest persisted canary summary. Scheduled health integration is gated off by default and only activates when the live worker sets HEALTH_AGENT_DISPATCH_CANARY_SCHEDULE=signals.
• infer({ timeout }) is an overall budget, not a per-fallback reset. Story 6 proved that reusing a fresh 10-minute timeout on every fallback attempt creates a hidden 30-minute failure chain (SIGTERM → exit 143) before the real story budget is exhausted. src/lib/inference.ts must spend the remaining deadline across attempts and preserve up to a one-hour explicit request budget for Restate PRD story runs.
• Timeout errors must say timeout, not exit 143: empty output. When pi is killed by the inference timer, surface pi timed out after <ms> in the thrown error and OTEL metadata so operators know it was our budget kill, not a mysterious subprocess crash.
• Do not import packages/cli/src/* from system-bus via relative paths. Keep runbook resolution local in packages/system-bus (or extract to a dedicated leaf package) and avoid creating @joelclaw/system-bus ↔ @joelclaw/sdk dependency cycles that break Turbo/Vercel.

Deploy: system-bus-worker (k8s)

~/Code/joelhooks/joelclaw/k8s/publish-system-bus-worker.sh
kubectl -n joelclaw rollout status deployment/system-bus-worker --timeout=180s
joelclaw refresh

Builds ARM64 image, pushes to GHCR, updates k8s deployment, verifies rollout.

Adding a Webhook Provider

See the webhooks skill for full details. Quick summary:

1. Create src/webhooks/providers/<service>.ts implementing WebhookProvider
2. Register in src/webhooks/server.ts
3. Add secret to WEBHOOK_SECRETS array in serve.ts
4. Store secret in agent-secrets: secrets add <service>_webhook_secret

Debugging

# Check worker health
curl http://localhost:3111/ | jq

# View registered functions
joelclaw functions

# Recent runs
joelclaw runs --count 20

# Inspect a specific run
joelclaw run <RUN_ID>

# Worker logs (k8s)
kubectl logs -n joelclaw deploy/system-bus-worker -f

# Inngest server logs
kubectl logs -n joelclaw inngest-0 | grep ERROR

# Force re-registration
curl -X PUT http://127.0.0.1:3111/api/inngest

Runtime forensics: stale RUNNING runs

When Inngest APIs disagree (runs list shows RUNNING, run detail shows terminal or non-cancellable state), treat it as runtime metadata drift, usually after SDK reachability failures.

Operational truths:

• Runtime DB is SQLite inside k8s Inngest pod: inngest-0:/data/main.db.
• trace_runs.status alone is not sufficient to infer terminality.
• Terminal source-of-truth is the presence of terminal history entries:
  • FunctionCompleted
  • FunctionFailed
  • FunctionCancelled

Safe reconciliation sequence:

1. Preview with joelclaw inngest sweep-stale-runs.
2. Apply with joelclaw inngest sweep-stale-runs --apply (auto backup + transactional writes).
3. If manual fallback is required:
   • Backup DB: kubectl -n joelclaw exec inngest-0 -- sqlite3 /data/main.db '.backup /data/main.db.pre-sweep-<ts>.sqlite'
   • Find stale candidates via trace_runs + function_finishes + history joins.
   • Insert missing terminal history (FunctionCancelled) for stale candidates.
   • Ensure function_finishes rows exist.
   • Update trace_runs.status to cancelled (500) only after history/finishes.
4. Verify with joelclaw run <id> and a fresh joelclaw runs --status RUNNING.

Key Files