{"slug":"strategy/business-strategy/L05_OPERATING_LAYER_WIRING_PLAN","title":"L0.5 Operating Layer Wiring Plan","section":"strategy","sectionLabel":"Strategy","status":"unclassified","type":"document","path":"strategy/business-strategy/L05_OPERATING_LAYER_WIRING_PLAN.mdx","bytes":12791,"description":null,"updated":null,"source":"integratewise/Integratewise-docs-wiki","content":"\n> **Created:** 2026-06-09  \n> **Status:** IN PROGRESS  \n> **Authority:** Nirmal (Founder) approval required for deployment\n\n---\n\n## Executive Summary\n\nThis document outlines the systematic wiring of the **L0.5 Operating Layer** (Cloudflare Durable Execution substrate) to transform IntegrateWise from a reactive application into an autonomous operating system.\n\n**L0.5 Components:**\n\n- **Cloudflare Workflows** → Durable, multi-step, stateful execution\n- **Cloudflare Queues** → Async processing with guaranteed delivery + DLQs\n- **Cron Triggers** → Scheduled, policy-bounded automation\n- **Durable Objects** → Stateful long-lived coordination (HITL gates, watchers)\n\n---\n\n## Current State Analysis (Updated 2026-06-09 Phase 2)\n\n### ✅ CONNECTED (Working)\n\n- `pipeline-process` queue → `integratewise-pipeline` consumer\n- `signals` queue → `integratewise-intelligence` consumer\n- `intelligence-events` queue → `integratewise-intelligence` consumer\n- `knowledge-ingest` queue → `integratewise-knowledge` consumer\n- `memory-consolidation` → `integratewise-knowledge` consumer\n- `ops-dlq` → `integratewise-intelligence` consumer\n- `connector-sync` queue → `connector-sync` consumer (webhook-driven + hourly cron)\n- **`intelligence-act` queue → producer binding added (Phase 2)** ✅\n\n### ⚠️ NEEDS MANUAL WIRING (Phase 2 Code Deployed)\n\n- **`continuity-triage` queue → consumer code deployed, needs Dashboard wiring** ⏳\n\n### ⚠️ CRON GAPS\n\n- `connector-sync`: Was 6 hours, needs webhook-driven (hourly backup)\n- `continuity`: Has handler, missing morning brief + backup crons\n- `intelligence`: No cron for morning brief or signal sweeps\n\n### ⚠️ MISSING WORKFLOWS\n\n- `TriageWorkflow` → exists in code, needs activation\n- `SyncWorkflow` → missing, needed for connector delta\n- `SignalWorkflow` → missing, needed for spine→signal automation\n- `MorningBriefWorkflow` → missing, needed for 7am brief\n- `ActWorkflow` → missing, needed for approved→execute\n\n---\n\n## TASK 1: FIX CRON TRIGGERS\n\n### 1.1 connector-sync: Hourly + Webhook-Driven\n\n**File:** `services/connector-sync/wrangler.toml`\n\n**Change:**\n\n```toml\n# OLD\n# crons = [\"0 */6 * * *\"]  # Every 6 hours\n\n# NEW\n[triggers]\ncrons = [\"0 * * * *\"]  # Every hour (backup sweep)\n# Real-time syncs triggered by Nango webhooks at /webhooks/nango\n```\n\n**Rationale:** Nango webhooks (`/webhooks/nango`) provide real-time sync triggers. Hourly cron is backup sweep for missed events.\n\n**Status:** ✅ DONE\n\n---\n\n### 1.2 continuity: Morning Brief + Daily Backup\n\n**File:** `services/continuity/wrangler.toml`\n\n**Change:**\n\n```toml\n# OLD\n[triggers]\ncrons = [\"0 */6 * * *\"]  # TTL cleanup only\n\n# NEW\n[triggers]\ncrons = [\n  \"0 7 * * *\",      # Morning brief at 7am\n  \"0 0 * * *\",      # Daily reindex + backup at midnight\n  \"0 */6 * * *\"     # TTL cleanup every 6 hours\n]\n```\n\n**Handler Logic:**\n\n```typescript\nasync scheduled(event: ScheduledEvent, env: Env): Promise<void> {\n  const cron = event.cron;\n\n  if (cron === \"0 7 * * *\") {\n    // Morning brief: trigger MorningBriefWorkflow\n    // TODO: Wire to WORKFLOW service binding\n  }\n  else if (cron === \"0 0 * * *\") {\n    // Daily backup: D1 → R2 + Vectorize reindex\n    // TODO: Replace with AI Search (PRIORITY 4)\n  }\n  else if (cron === \"0 */6 * * *\") {\n    // TTL cleanup (existing)\n    await handleScheduled(event, env);\n  }\n}\n```\n\n**Status:** ✅ DONE (wrangler.toml), ⏳ PENDING (handler impl)\n\n---\n\n### 1.3 intelligence: Morning Brief + Signal Sweep\n\n**File:** `services/intelligence/wrangler.toml`\n\n**Change:**\n\n```toml\n# NEW\n[triggers]\ncrons = [\n  \"0 7 * * *\",      # Morning brief at 7am\n  \"0 */6 * * *\"     # Signal analysis sweep every 6 hours\n]\n```\n\n**Status:** ✅ DONE\n\n---\n\n## TASK 2: FIX QUEUE WIRING\n\n### 2.1 continuity-triage Queue\n\n**Problem:** Producers exist (20), no consumer wired\n\n**Solution:** Add consumer to `continuity` worker\n\n**File:** `services/continuity/wrangler.toml`\n\n```toml\n[[queues.consumers]]\nqueue = \"continuity-triage\"\nmax_batch_size = 10\nmax_batch_timeout = 30\n```\n\n**Status:** ⏳ PENDING\n\n---\n\n### 2.2 intelligence-act Queue\n\n**Problem:** Consumer exists (intelligence), nothing produces to it\n\n**Solution:** Wire Act service to publish to queue after HITL approval\n\n**File:** `services/intelligence/src/act/*.ts`\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**Status:** ⏳ PENDING (needs queue producer binding)\n\n---\n\n### 2.3 connector-sync Queue\n\n**Problem:** Consumer exists, nothing produces\n\n**Solution:** Nango webhook handler triggers sync jobs\n\n**File:** `services/webhook-ingress/src/index.ts`\n\nAlready wired! Nango webhook handler sends to connector-sync queue:\n\n```typescript\nif (eventType === \"connection.created\" || eventType === \"sync.success\") {\n  await env.CONNECTOR_SYNC_QUEUE.send({\n    tenant_id: tenantId,\n    provider: provider,\n    sync_type: \"delta\",\n    triggered_by: \"webhook\",\n  });\n}\n```\n\n**Status:** ✅ ALREADY WIRED (verify producer binding exists)\n\n---\n\n### 2.4 execution-outcomes Queue\n\n**Problem:** Completely disconnected (0 producers, 0 consumers)\n\n**Decision:**\n\n- **Option A:** Delete the queue (unused)\n- **Option B:** Wire to Act service for execution tracking\n\n**Recommendation:** DELETE - execution tracking happens via D1 `action_executions` table\n\n**Status:** ⏳ PENDING DECISION\n\n---\n\n## TASK 3: CREATE AI SEARCH INSTANCE\n\n### 3.1 Current L4 Library Architecture\n\n**Current Flow:**\n\n```\nKnowledge content → R2 bucket → Manual vectorization → Vectorize index → D1 cache\n```\n\n**Problems:**\n\n- Manual indexing required\n- No built-in search API\n- No MCP endpoint\n- No chat completions interface\n\n---\n\n### 3.2 AI Search Migration\n\n**New Flow:**\n\n```\nKnowledge content → AI Search index → /search API + /chat/completions + MCP\n```\n\n**Benefits:**\n\n- Managed indexing (auto-sync from R2/D1)\n- Built-in search and chat APIs\n- MCP-compatible endpoint\n- Versioning and rollback\n- Hybrid search (vector + keyword)\n\n---\n\n### 3.3 Implementation Steps\n\n#### Step 1: Create AI Search Index\n\n```bash\nwrangler ai-search create integratewise-knowledge \\\n  --namespace=knowledge \\\n  --description=\"IntegrateWise L4 Library - org_memory + KB\" \\\n  --embedding-model=@cf/baai/bge-base-en-v1.5\n```\n\n#### Step 2: Wire to Knowledge Worker\n\n**File:** `services/knowledge/wrangler.toml`\n\n```toml\n[[ai_search_namespaces]]\nbinding = \"AI_SEARCH\"\nnamespace = \"integratewise-knowledge\"\n```\n\n#### Step 3: Migrate Content\n\n```typescript\n// Bulk upload from existing org_memory\nconst documents = await fetchOrgMemory();\nawait env.AI_SEARCH.upsert(\n  documents.map((doc) => ({\n    id: doc.id,\n    text: doc.content_md,\n    metadata: {\n      title: doc.title,\n      tags: doc.tags,\n      topics: doc.topics,\n      governance_state: doc.governance_state,\n    },\n  }))\n);\n```\n\n#### Step 4: Update Search Endpoints\n\n**File:** `services/knowledge/src/index.ts`\n\n```typescript\n// OLD: Manual Vectorize query\napp.get(\"/search\", async (c) => {\n  const results = await vectorizeQuery(c.env.VECTOR_INDEX, query);\n  // ...\n});\n\n// NEW: AI Search API\napp.get(\"/search\", async (c) => {\n  const results = await c.env.AI_SEARCH.search(query, {\n    limit: 10,\n    includeVectors: false,\n  });\n  // ...\n});\n```\n\n**Status:** ⏳ PENDING (AI Search in beta, verify availability)\n\n---\n\n## TASK 4: DEPLOY MISSING WORKFLOWS\n\n### 4.1 TriageWorkflow\n\n**Status:** ✅ EXISTS in `services/intelligence/src/workflows/triage-workflow.ts`\n\n**Binding:** ✅ ALREADY CONFIGURED in `intelligence/wrangler.toml`\n\n**Action:** Verify it's being invoked correctly\n\n---\n\n### 4.2 Missing Workflows\n\n#### MorningBriefWorkflow\n\n- **Purpose:** Generate 7am brief for /desk\n- **Triggers:** 7am cron (continuity, intelligence)\n- **Steps:**\n  1. Fetch overnight signals\n  2. Query Entity 360 changes\n  3. Generate brief via AI\n  4. Store in D1 cache\n  5. Broadcast to connected clients\n\n**Status:** ⏳ TODO\n\n---\n\n#### SignalWorkflow\n\n- **Purpose:** Spine change → signal generation → broadcast\n- **Triggers:** Pipeline writes, hourly sweep\n- **Steps:**\n  1. Detect spine changes\n  2. Run signal analyzer\n  3. Create signal records\n  4. Broadcast to workspaces\n\n**Status:** ⏳ TODO\n\n---\n\n#### ActWorkflow\n\n- **Purpose:** Approved proposal → execute action\n- **Triggers:** HITL approval, intelligence-act queue\n- **Steps:**\n  1. Fetch approved proposal\n  2. Resolve target tool\n  3. Execute via connector\n  4. Track outcome\n  5. Update proposal status\n\n**Status:** ⏳ TODO\n\n---\n\n## Deployment Sequence\n\n### Phase 1: Cron Fixes (Safe, No Breaking Changes) — ✅ DEPLOYED\n\n1. ✅ connector-sync: hourly cron (wrangler.toml)\n2. ✅ continuity: +7am, +midnight crons (wrangler.toml)\n3. ✅ intelligence: +7am, +6h crons (wrangler.toml)\n4. ✅ Implement continuity scheduled handler\n5. ✅ Implement intelligence scheduled handler\n6. ✅ DEPLOYED 2026-06-09 (manual cron wiring required via Dashboard)\n\n### Phase 2: Queue Wiring (Requires Testing) — ✅ DEPLOYED\n\n1. ✅ Add continuity-triage consumer (wrangler.toml + handler)\n2. ✅ Wire intelligence-act producer binding\n3. ✅ Verify connector-sync webhook flow (already wired)\n4. ✅ Delete execution-outcomes queue (done)\n5. ✅ DEPLOYED 2026-06-09 (manual queue consumer wiring required via Dashboard)\n\n**See:** `L05_PHASE2_DEPLOYMENT_SUMMARY.md` for full deployment details.\n\n### Phase 3: AI Search (Requires Beta Access) — ⏳ NEXT\n\n1. Add continuity-triage consumer\n2. Wire intelligence-act producer\n3. Verify connector-sync webhook flow\n4. Delete execution-outcomes queue\n5. Deploy + test each queue independently\n\n### Phase 3: AI Search (Requires Beta Access)\n\n1. Create AI Search index\n2. Bulk upload org_memory content\n3. Update search endpoints\n4. Test MCP integration\n5. Deprecate manual Vectorize flow\n\n### Phase 4: Workflows (Requires Development)\n\n1. Implement MorningBriefWorkflow\n2. Implement SignalWorkflow\n3. Implement ActWorkflow\n4. Wire cron triggers\n5. Test durable execution + retries\n\n---\n\n## Verification Commands\n\n### Check Deployed Crons\n\n```bash\nwrangler deployments list --name integratewise-connector-sync\nwrangler deployments list --name continuity\nwrangler deployments list --name intelligence\n```\n\n### List Queues\n\n```bash\nwrangler queues list\n```\n\n### Check Queue Consumers\n\n```bash\nwrangler queues consumer list continuity-triage\nwrangler queues consumer list intelligence-act\n```\n\n### Test Webhook Flow\n\n```bash\ncurl -X POST 'https://ingress.dev.integratewise.ai/webhooks/nango' \\\n  -H 'Content-Type: application/json' \\\n  -d '{\"type\":\"sync.success\",\"connectionId\":\"iw-customer-zero\",\"provider\":\"hubspot\"}'\n```\n\n---\n\n## Success Criteria\n\n### L0.5 Fully Operational When:\n\n- ✅ All cron triggers deployed and running\n- ✅ All queue consumers wired with producers\n- ✅ Workflows executing durably with retries\n- ✅ AI Search replacing manual vectorization\n- ✅ Morning brief generated daily at 7am\n- ✅ Signal analysis sweep running every 6 hours\n- ✅ Connector sync webhook-driven with hourly backup\n- ✅ Act workflow executing approved proposals\n\n---\n\n## Next Actions\n\n**PHASE 1 COMPLETE — READY FOR DEPLOYMENT:**\n\nAll Phase 1 code changes are complete:\n\n1. ✅ Cron triggers added to wrangler.toml files (connector-sync, continuity, intelligence)\n2. ✅ Continuity scheduled handler implemented (7am brief, midnight backup, 6h cleanup)\n3. ✅ Intelligence scheduled handler implemented (7am brief, 6h signal sweep)\n\n**AWAITING NIRMAL APPROVAL TO DEPLOY:**\n\n```bash\n# Unset Cloudflare API token (critical - see AGENTS.md Hard Rules)\nunset CLOUDFLARE_API_TOKEN\n\n# Deploy Phase 1 changes\nwrangler deploy --env=\"\" services/connector-sync\nwrangler deploy --env=\"\" services/continuity\nwrangler deploy --env=\"\" services/intelligence\n```\n\n**AFTER PHASE 1 DEPLOYMENT:**\n\n1. Monitor cron triggers in Cloudflare dashboard\n2. Verify scheduled handlers execute correctly\n3. Proceed to Phase 2: Queue Wiring\n\n**PHASE 2 QUEUE WIRING (After Phase 1 approval):**\n\n1. Add continuity-triage consumer\n2. Wire intelligence-act producer\n3. Verify connector-sync webhook flow\n4. Delete execution-outcomes queue\n\n**PHASE 3 AI SEARCH (After Phase 2):**\n\n1. Request AI Search beta access\n2. Create AI Search index\n3. Bulk upload org_memory content\n4. Update search endpoints\n\n**PHASE 4 WORKFLOWS (After Phase 3):**\n\n1. Develop MorningBriefWorkflow\n2. Develop SignalWorkflow\n3. Develop ActWorkflow\n4. Wire workflow triggers\n\n---\n\n**Document Status:** Phase 1 READY FOR DEPLOYMENT  \n**Last Updated:** 2026-06-09T13:15:00Z  \n**Next Review:** After Phase 1 deployment + monitoring\n"}