{"slug":"operations/L05_PHASE2_QUEUE_WIRING","title":"L0.5 Phase 2: Queue Wiring","section":"operations","sectionLabel":"Operations","status":"unclassified","type":"document","path":"operations/L05_PHASE2_QUEUE_WIRING.mdx","bytes":11400,"description":null,"updated":null,"source":"integratewise/Integratewise-docs-wiki","content":"\n> **Date:** 2026-06-09  \n> **Status:** READY FOR DEPLOYMENT  \n> **Phase:** 2 of 4  \n> **Previous:** Phase 1 (Cron Triggers) — Workers deployed, manual cron wiring required\n\n---\n\n## Executive Summary\n\nPhase 2 completes the **queue wiring** for the L0.5 Operating Layer, ensuring all queue producers and consumers are correctly connected for async processing and durability.\n\n**Changes:**\n\n1. ✅ Added `continuity-triage` queue consumer to `continuity` worker\n2. ✅ Added `continuity-triage` queue handler implementation\n3. ✅ Added `intelligence-act` queue producer binding to `intelligence` worker\n4. ⏳ Queue producer implementation (future — when HITL approval logic is ready)\n\n---\n\n## Changes Made\n\n### 1. continuity-triage Queue Consumer\n\n**Problem:** Producers exist (20+ calls to `TASKS_QUEUE.send`), but no consumer wired.\n\n**Solution:** Added consumer + handler to `continuity` worker.\n\n#### File: `services/continuity/wrangler.toml`\n\n**Added:**\n\n```toml\n# Queue consumer for continuity-triage\n[[queues.consumers]]\nqueue = \"continuity-triage\"\nmax_batch_size = 10\nmax_batch_timeout = 30\n```\n\n#### File: `services/continuity/src/index.ts`\n\n**Added `queue()` export:**\n\n```typescript\nexport async function queue(\n  batch: MessageBatch<{\n    type: string;\n    tenant_id: string;\n    session_id: string;\n    event_count?: number;\n  }>,\n  env: Env\n): Promise<void> {\n  console.log(`[Continuity Queue] Processing ${batch.messages.length} messages`);\n\n  for (const message of batch.messages) {\n    try {\n      const { type, tenant_id, session_id, event_count } = message.body;\n\n      if (type === \"session_ready_for_triage\") {\n        await handleTriageRequest(env, tenant_id, session_id, event_count);\n        message.ack();\n      } else {\n        console.warn(`[Continuity Queue] Unknown message type: ${type}`);\n        message.ack(); // Ack unknown types to prevent retry\n      }\n    } catch (err: any) {\n      console.error(`[Continuity Queue] Message processing failed:`, err.message);\n      message.retry();\n    }\n  }\n}\n```\n\n**Added `handleTriageRequest()` helper:**\n\n```typescript\nasync function handleTriageRequest(\n  env: Env,\n  tenantId: string,\n  sessionId: string,\n  eventCount?: number\n): Promise<void> {\n  console.log(`[Continuity Triage] Processing session ${sessionId} (${eventCount} events)`);\n\n  // Fetch session + events\n  const { data: session } = await selectOne<SessionContainer>(db, \"memory.session_containers\", {\n    id: sessionId,\n    tenant_id: tenantId,\n  });\n\n  const { data: events } = await select<SessionEvent>(db, \"memory.session_events\", {\n    session_id: `eq.${sessionId}`,\n    tenant_id: `eq.${tenantId}`,\n    order: \"created_at.asc\",\n    limit: \"1000\",\n  });\n\n  // Auto-triage logic (placeholder — replace with AI analysis)\n  const shouldPromote = events.length > 50; // Simple rule for now\n\n  if (shouldPromote) {\n    // Promote to knowledge\n    const decision: TriageDecision = {\n      session_id: sessionId,\n      outcome: \"promote_to_knowledge\",\n      reason: \"Auto-promoted: high-value session with significant event count\",\n      approved_by: \"triage-bot\",\n    };\n\n    await promoteSessionToKnowledge(env, tenantId, sessionId, decision);\n\n    await update(\n      db,\n      \"memory.session_containers\",\n      { id: sessionId },\n      {\n        triage_status: \"approved\",\n        triage_outcome: \"promote_to_knowledge\",\n        updated_at: nowIso(),\n      }\n    );\n  } else {\n    // Retain as ephemeral\n    await update(\n      db,\n      \"memory.session_containers\",\n      { id: sessionId },\n      {\n        triage_status: \"approved\",\n        triage_outcome: \"retain_ephemeral\",\n        updated_at: nowIso(),\n      }\n    );\n  }\n}\n```\n\n---\n\n### 2. intelligence-act Queue Producer\n\n**Problem:** Consumer exists (intelligence worker), but nothing produces to it.\n\n**Solution:** Added producer binding (code implementation deferred to future).\n\n#### File: `services/intelligence/wrangler.toml`\n\n**Added to all 3 environments (dev, test, prod):**\n\n```toml\n# --- Queue Producers (dev) ---\n[[queues.producers]]\nqueue = \"intelligence-act\"\nbinding = \"INTELLIGENCE_ACT_QUEUE\"\n```\n\n**Code Implementation (Future):**\n\nWhen HITL approval logic is implemented, add:\n\n```typescript\n// After HITL approval in govern flow:\nawait env.INTELLIGENCE_ACT_QUEUE.send({\n  action_id: proposal.id,\n  action_type: proposal.action_type,\n  tenant_id: proposal.tenant_id,\n  approved_by: approval.user_id,\n  approved_at: new Date().toISOString(),\n});\n```\n\n**Current Status:** Binding ready, producer code deferred until HITL approval endpoints are built.\n\n---\n\n### 3. connector-sync Queue\n\n**Status:** ✅ ALREADY WIRED\n\n**Verification Needed:**\n\n1. Check `webhook-ingress` produces to queue on Nango webhooks\n2. Check `connector-sync` consumes from queue\n3. Verify hourly cron backup sweep\n\n**Files:**\n\n- `services/webhook-ingress/src/index.ts` (producer)\n- `services/connector-sync/src/index.ts` (consumer)\n- `services/connector-sync/wrangler.toml` (hourly cron)\n\n---\n\n### 4. execution-outcomes Queue\n\n**Decision:** DELETE (unused)\n\n**Rationale:**\n\n- 0 producers\n- 0 consumers\n- Execution tracking happens via D1 `action_executions` table\n- No business logic references this queue\n\n**Command:**\n\n```bash\nwrangler queues delete execution-outcomes\n```\n\n---\n\n## Deployment Steps\n\n### Step 1: Deploy continuity Worker\n\n```bash\ncd /Users/nirmal/Github/integratewise-live\nunset CLOUDFLARE_API_TOKEN\nwrangler deploy --env=\"\" services/continuity\n```\n\n**Expected Output:**\n\n- ✅ Worker deployed\n- ⚠️ Cron schedules may fail (API issue from Phase 1)\n- ⚠️ Queue consumer wiring may fail (API issue from Phase 1)\n\n**Manual Wiring (if needed):**\n\n- Cloudflare Dashboard → Queues → `continuity-triage` → Add Consumer → `continuity`\n\n---\n\n### Step 2: Deploy intelligence Worker\n\n```bash\nwrangler deploy --env=\"\" services/intelligence\n```\n\n**Expected Output:**\n\n- ✅ Worker deployed\n- ✅ Producer binding added (no API call needed for producers)\n- ⚠️ Consumer wiring may fail (Phase 1 issue persists)\n\n---\n\n### Step 3: Delete execution-outcomes Queue\n\n```bash\nwrangler queues delete execution-outcomes\n```\n\n**Expected Output:**\n\n- ✅ Queue deleted (no consumers, safe)\n\n---\n\n### Step 4: Verify Deployments\n\n#### Check Worker Versions\n\n```bash\nwrangler deployments list --name continuity\nwrangler deployments list --name intelligence\n```\n\n#### Tail Logs\n\n```bash\nwrangler tail continuity\nwrangler tail intelligence\n```\n\n#### Check Queue Consumer Status\n\n```bash\nwrangler queues consumer list continuity-triage\nwrangler queues consumer list intelligence-act\n```\n\n#### Test continuity-triage Queue\n\n```bash\n# Send test message\ncurl -X POST 'https://continuity.connect-a1b.workers.dev/v1/continuity/sessions/<session_id>/close' \\\n  -H 'x-tenant-id: iw-customer-zero' \\\n  -H 'Content-Type: application/json'\n\n# Check logs\nwrangler tail continuity\n```\n\n---\n\n## Success Criteria\n\n### Phase 2 Complete When:\n\n- ✅ `continuity-triage` queue has consumer wired\n- ✅ `continuity-triage` queue handler processes messages correctly\n- ✅ `intelligence-act` queue producer binding exists\n- ✅ `execution-outcomes` queue deleted\n- ✅ All workers deployed with latest code\n- ✅ Logs show queue processing working\n\n---\n\n## Queue Status Summary (After Phase 2)\n\n| Queue Name               | Producer                   | Consumer             | Status         |\n| ------------------------ | -------------------------- | -------------------- | -------------- |\n| `pipeline-process`       | webhook-ingress, connector | pipeline             | ✅ WORKING     |\n| `signals`                | pipeline                   | intelligence         | ✅ WORKING     |\n| `intelligence-events`    | pipeline, connector        | intelligence         | ✅ WORKING     |\n| `knowledge-ingest`       | pipeline, connector        | knowledge            | ✅ WORKING     |\n| `memory-consolidation`   | pipeline, knowledge        | knowledge            | ✅ WORKING     |\n| `ops-dlq`                | pipeline, intelligence     | intelligence         | ✅ WORKING     |\n| **`continuity-triage`**  | continuity (20+ calls)     | **continuity (NEW)** | **✅ WIRED**   |\n| **`intelligence-act`**   | (future HITL approval)     | intelligence         | **✅ READY**   |\n| **`connector-sync`**     | webhook-ingress (Nango)    | connector-sync       | **⚠️ VERIFY**  |\n| ~~`execution-outcomes`~~ | (none)                     | (none)               | **🗑️ DELETED** |\n\n---\n\n## Next Steps\n\n### Immediate (After Deployment)\n\n1. **Manual Queue Wiring (if Cloudflare API fails):**\n   - Add `continuity-triage` consumer via Dashboard\n   - Add 4 intelligence queue consumers (Phase 1 carryover)\n\n2. **Verify Queue Flows:**\n   - Test continuity session close → triage queue → handler\n   - Test webhook → connector-sync queue flow\n   - Monitor logs for errors\n\n3. **Update Documentation:**\n   - Update `docs/operations/CF_WIRE_STATUS.md` with queue status\n   - Update `L05_OPERATING_LAYER_WIRING_PLAN.md` with Phase 2 completion\n\n---\n\n### Phase 3: AI Search Migration\n\n1. Request AI Search beta access (if not already available)\n2. Create AI Search namespace: `integratewise-knowledge`\n3. Bulk upload org_memory content\n4. Update search endpoints in knowledge service\n5. Deprecate manual Vectorize flow\n\n---\n\n### Phase 4: Workflow Development\n\n1. Implement `MorningBriefWorkflow` (durable multi-step execution)\n2. Implement `SignalWorkflow` (spine change → signal generation)\n3. Implement `ActWorkflow` (approved proposal → execute action)\n4. Wire workflow triggers to crons + queues\n\n---\n\n## Files Modified\n\n### Phase 2 Changes\n\n1. **`services/continuity/wrangler.toml`**\n   - Added `[[queues.consumers]]` for `continuity-triage`\n\n2. **`services/continuity/src/index.ts`**\n   - Added `queue()` export\n   - Added `handleTriageRequest()` helper\n   - Wired auto-triage logic (placeholder for AI)\n\n3. **`services/intelligence/wrangler.toml`**\n   - Added `[[queues.producers]]` for `intelligence-act` (all 3 envs)\n\n---\n\n## Risks & Mitigations\n\n| Risk                                         | Impact                 | Mitigation                                                      |\n| -------------------------------------------- | ---------------------- | --------------------------------------------------------------- |\n| Cloudflare API `/queues/.../consumers` fails | Manual wiring required | Dashboard fallback documented                                   |\n| Triage logic promotes all sessions           | Knowledge pollution    | Start with conservative rule (>50 events), replace with AI soon |\n| intelligence-act producer not implemented    | Queue stays empty      | Defer to HITL approval endpoint development                     |\n| connector-sync not producing                 | Sync delays            | Verify Nango webhook handler + hourly cron backup               |\n\n---\n\n## Rollback Plan\n\nIf Phase 2 deployment causes issues:\n\n```bash\n# Rollback continuity\nwrangler rollback --name continuity\n\n# Rollback intelligence\nwrangler rollback --name intelligence\n\n# Recreate execution-outcomes queue (if needed)\nwrangler queues create execution-outcomes\n```\n\n---\n\n**Phase 2 Status:** READY FOR DEPLOYMENT  \n**Approval Required:** Nirmal  \n**Deployment Time:** ~10 minutes  \n**Breaking Changes:** None (additive only)\n\n**Last Updated:** 2026-06-09T14:45:00Z\n"}