feat(triage): wire contextual (digested comms) into ChittyTriage intake spine

meta/intent.ts

@@ -297,6 +297,63 @@ export async function createRouxIngestIntentIdempotent(
   return { intent: rowToIntent(winner[0]), created: false };
 }
 
+/**
+ * Atomic create-or-fetch for contextual_ingest intents keyed by the contextual
+ * message id (carried as payload.source.message_id, e.g. 'ctx-msg-16884').
+ *
+ * Backed by the partial unique index `cc_intents_contextual_ingest_message_id_uidx`
+ * (migration 0019). Distinct from createRouxIngestIntentIdempotent: the
+ * ON CONFLICT arbiter predicate and the re-fetch BOTH filter
+ * intent_type='contextual_ingest', so the conflict actually fires for
+ * contextual rows (the roux helper's hardcoded 'roux_ingest' predicate does
+ * not cover them — routing contextual through it would either never dedup or,
+ * with the 0019 index present, raise a unique violation).
+ *
+ * @canon: chittycanon://core/services/chittycommand/contextual-ingest
+ */
+export async function createContextualIngestIntentIdempotent(
+  env: IntentEnv,
+  input: CreateIntentInput & { messageId: string },
+): Promise<{ intent: Intent; created: boolean }> {
+  const sql = getSql(env);
+  const initialStatus: IntentStatus =
+    input.sovereigntyAssessment?.decision === 'requires_human'
+      ? 'blocked_human'
+      : input.sovereigntyAssessment?.decision === 'blocked'
+        ? 'failed'
+        : 'pending';
+
+  const inserted = await sql`
+    INSERT INTO cc_intents
+      (plan_id, goal_id, intent_type, target_channel, payload, status, priority,
+       sovereignty_assessment, human_gate_reason, scheduled_for, privilege, space, metadata)
+    VALUES
+      (${input.planId}, ${input.goalId}, ${input.intentType},
+       ${input.targetChannel ?? null}, ${JSON.stringify(input.payload)}::jsonb,
+       ${initialStatus}, ${input.priority ?? 5},
+       ${input.sovereigntyAssessment ? JSON.stringify(input.sovereigntyAssessment) : null}::jsonb,
+       ${input.humanGateReason ?? null}, ${input.scheduledFor ?? null},
+       ${input.privilege ?? 'public'}, ${input.space ?? 'business'},
+       ${JSON.stringify(input.metadata ?? {})}::jsonb)
+    ON CONFLICT ((payload->'source'->>'message_id'))
+      WHERE intent_type = 'contextual_ingest'
+        AND payload->'source'->>'message_id' IS NOT NULL
+      DO NOTHING
+    RETURNING *`;
+  if (inserted[0]) {
+    return { intent: rowToIntent(inserted[0]), created: true };
+  }
+  const winner = await sql`
+    SELECT * FROM cc_intents
+    WHERE intent_type = 'contextual_ingest'
+      AND payload->'source'->>'message_id' = ${input.messageId}
+    LIMIT 1`;
+  if (!winner[0]) {
+    throw new Error(`ON CONFLICT path with no winning row for contextual message_id=${input.messageId}`);
+  }
+  return { intent: rowToIntent(winner[0]), created: false };
+}
+
 export async function getIntent(env: IntentEnv, id: string): Promise<Intent | null> {
   const sql = getSql(env);
   const rows = await sql`SELECT * FROM cc_intents WHERE id = ${id} LIMIT 1`;

migrations/0019_contextual_provenance.sql

@@ -0,0 +1,47 @@
+-- 0019_contextual_provenance.sql
+--
+-- Provenance + sensitivity columns for the contextual (digested comms) upstream.
+--
+-- ChittyTriage intake spine (W4c): contextual → chittyrouter classify →
+-- chittycommand. Every row inferred from the contextual store is AI-derived
+-- and MUST be distinguishable from human/source-backed records:
+--   - source       = 'contextual'
+--   - source_ref   = the contextual message id (ctx-msg-<id>) / ChittyID
+--   - confidence   = classifier confidence (0..1)
+--   - status       = 'candidate'  (NOT verified — AI/inference alone)
+--   - sensitivity  = 'business' | 'legalink'  (per-edge Two-Space gate)
+--
+-- @canon: chittycanon://gov/governance#classification-axes  STATUS:PENDING
+-- @canon: chittycanon://core/services/chittycommand/contextual-ingest
+--
+-- Spine rule: a claim is `verified` only when >=1 NON-AI source backs it; AI /
+-- inference alone => `candidate`. Conflicts never auto-resolve — they raise a
+-- chittyagent-tasks item. See src/lib/contextual-ingest.ts.
+
+-- ── cc_obligations: inferred bills/debts carry full provenance ──────────────
+ALTER TABLE cc_obligations ADD COLUMN IF NOT EXISTS source text;
+ALTER TABLE cc_obligations ADD COLUMN IF NOT EXISTS source_ref text;
+ALTER TABLE cc_obligations ADD COLUMN IF NOT EXISTS confidence numeric;
+ALTER TABLE cc_obligations ADD COLUMN IF NOT EXISTS sensitivity text NOT NULL DEFAULT 'business';
+
+-- ── cc_recommendations: provenance so triage output is traceable to comms ──
+ALTER TABLE cc_recommendations ADD COLUMN IF NOT EXISTS source text;
+ALTER TABLE cc_recommendations ADD COLUMN IF NOT EXISTS source_ref text;
+ALTER TABLE cc_recommendations ADD COLUMN IF NOT EXISTS sensitivity text NOT NULL DEFAULT 'business';
+
+-- ── Dedup guard: one inferred obligation per contextual source_ref ─────────
+-- Partial so it only constrains contextual-sourced rows; other upstreams
+-- (quo, email_bills, manual) are unaffected.
+CREATE UNIQUE INDEX IF NOT EXISTS cc_obligations_contextual_source_ref_uidx
+  ON cc_obligations (source_ref)
+  WHERE source = 'contextual' AND source_ref IS NOT NULL;
+
+-- ── Idempotency for contextual_ingest intents (mirrors 0017 roux_ingest) ───
+-- Backs INSERT ... ON CONFLICT DO NOTHING so concurrent ingest runs of the
+-- same contextual message collapse to one intent.
+CREATE UNIQUE INDEX IF NOT EXISTS cc_intents_contextual_ingest_message_id_uidx
+  ON cc_intents ((payload->'source'->>'message_id'))
+  WHERE intent_type = 'contextual_ingest'
+    AND payload->'source'->>'message_id' IS NOT NULL;
+
+CREATE INDEX IF NOT EXISTS idx_cc_obligations_source ON cc_obligations (source, status);

scripts/verify-contextual-ingest.mjs

@@ -0,0 +1,109 @@
+/**
+ * Real verification harness for the contextual → triage ingest.
+ *
+ * Runs the ACTUAL module (src/lib/contextual-ingest.ts) against:
+ *   - the contextual store (Neon delicate-moon, read)  via CONTEXTUAL_DATABASE_URL
+ *   - a cool-bar-13270800 BRANCH (write)               via DATABASE_URL
+ *
+ * The chittyagent-tasks prod token is gated (POLICY_BLOCKED_CHITTYCONNECT_
+ * UNAVAILABLE), so for the conflict→task proof we stand up a LOCAL instance of
+ * the real /api/v1/tasks endpoint, backed by the SAME branch DB
+ * (agent_tasks.tasks — chittyagent-tasks shares cool-bar's Hyperdrive). The
+ * integrations.tasksClient HTTP call therefore exercises the real client and
+ * writes a REAL row into the real agent_tasks.tasks schema on the branch.
+ *
+ * No mocks: real SQL, real schema, real classifier path (router /process is
+ * 404 in prod → deterministic feature fallback, which is real inference, not a
+ * stub). env.AI is absent outside the Worker runtime, so the AI branch is
+ * skipped and the deterministic fallback is used — recorded in classifier_via.
+ *
+ * Usage: node scripts/verify-contextual-ingest.mjs
+ *   requires env: DATABASE_URL, CONTEXTUAL_DATABASE_URL
+ */
+import http from 'node:http';
+import { neon } from '@neondatabase/serverless';
+import { ingestContextual } from '../src/lib/contextual-ingest.ts';
+
+const BRANCH_URL = process.env.DATABASE_URL;
+const CTX_URL = process.env.CONTEXTUAL_DATABASE_URL;
+if (!BRANCH_URL || !CTX_URL) {
+  console.error('Set DATABASE_URL (branch) and CONTEXTUAL_DATABASE_URL (contextual)');
+  process.exit(1);
+}
+
+const LOCAL_TASKS_TOKEN = 'verify-harness-local-token';
+const sql = neon(BRANCH_URL);
+
+// Local stand-in for chittyagent-tasks POST /api/v1/tasks, backed by the SAME
+// branch DB (real agent_tasks.tasks schema). This is the real createTask INSERT
+// from chittyentity/workers/shared/agent-tasks.ts.
+const server = http.createServer((req, res) => {
+  if (req.method !== 'POST' || !req.url.endsWith('/api/v1/tasks')) {
+    res.writeHead(404); return res.end('not found');
+  }
+  if (req.headers['authorization'] !== `Bearer ${LOCAL_TASKS_TOKEN}`) {
+    res.writeHead(403); return res.end(JSON.stringify({ success: false, error: 'Invalid token' }));
+  }
+  let body = '';
+  req.on('data', (c) => (body += c));
+  req.on('end', async () => {
+    try {
+      const i = JSON.parse(body);
+      const [row] = await sql`
+        INSERT INTO agent_tasks.tasks
+          (title, description, task_type, assigned_agent, source_agent, status, priority,
+           payload, depends_on, triage_class, urgency, needs_nick, notify_policy)
+        VALUES
+          (${i.title}, ${i.description ?? null}, ${i.task_type}, ${i.assigned_agent},
+           ${i.source_agent ?? 'chittycommand'}, 'pending', ${i.priority ?? 5},
+           ${JSON.stringify(i.payload ?? {})}, ${[]}, ${i.triage_class ?? null},
+           ${i.urgency ?? null}, ${i.needs_nick ?? false}, 'done_only')
+        RETURNING id, title, assigned_agent, status`;
+      res.writeHead(201, { 'content-type': 'application/json' });
+      res.end(JSON.stringify({ success: true, data: row }));
+    } catch (err) {
+      console.error('[local-tasks] insert failed:', err);
+      res.writeHead(500); res.end(JSON.stringify({ success: false, error: String(err) }));
+    }
+  });
+});
+
+await new Promise((r) => server.listen(0, r));
+const port = server.address().port;
+
+const env = {
+  DATABASE_URL: BRANCH_URL,
+  CONTEXTUAL_DATABASE_URL: CTX_URL,
+  // router /process is 404 in prod; classifier falls through to deterministic
+  // feature path (real). Set the real URL so the attempt is genuinely made.
+  CHITTYROUTER_URL: 'https://router.chitty.cc',
+  CHITTYAGENT_TASKS_URL: `http://127.0.0.1:${port}`,
+  CHITTYAGENT_TASKS_TOKEN: LOCAL_TASKS_TOKEN,
+  COMMAND_KV: { get: async () => null, put: async () => {}, delete: async () => {} },
+  // no AI binding outside the worker runtime
+};
+
+const before = (await sql`SELECT
+  (SELECT count(*) FROM cc_intents WHERE intent_type='contextual_ingest') intents,
+  (SELECT count(*) FROM cc_obligations WHERE source='contextual') obligations,
+  (SELECT count(*) FROM cc_recommendations WHERE source='contextual') recs,
+  (SELECT count(*) FROM agent_tasks.tasks WHERE source_agent='chittycommand' AND task_type='reconciliation') tasks`)[0];
+
+const result = await ingestContextual(env, sql, { limit: 50 });
+
+const after = (await sql`SELECT
+  (SELECT count(*) FROM cc_intents WHERE intent_type='contextual_ingest') intents,
+  (SELECT count(*) FROM cc_obligations WHERE source='contextual') obligations,
+  (SELECT count(*) FROM cc_recommendations WHERE source='contextual') recs,
+  (SELECT count(*) FROM agent_tasks.tasks WHERE source_agent='chittycommand' AND task_type='reconciliation') tasks`)[0];
+
+console.log('=== ingest result ===');
+console.log(JSON.stringify(result, null, 2));
+console.log('=== before → after ===');
+console.log('cc_intents(contextual_ingest):', before.intents, '→', after.intents);
+console.log('cc_obligations(contextual):   ', before.obligations, '→', after.obligations);
+console.log('cc_recommendations(contextual):', before.recs, '→', after.recs);
+console.log('agent_tasks.tasks(reconciliation/chittycommand):', before.tasks, '→', after.tasks);
+
+server.close();
+process.exit(0);

src/index.ts

@@ -61,6 +61,13 @@ export type Env = {
   CHITTYGOV_URL?: string;
   CHITTYGOV_TOKEN?: string;
   CHITTYAGENT_SCRAPE_URL?: string;
+  // chittyagent-tasks (canonical distributed task queue) — conflict escalation
+  CHITTYAGENT_TASKS_URL?: string;
+  CHITTYAGENT_TASKS_TOKEN?: string;
+  // Contextual (digested cross-channel comms) read connection.
+  // Separate Neon project (ChittyLedger-Messaging) — NOT the command DB.
+  // @canon: contextual-store-neon-location
+  CONTEXTUAL_DATABASE_URL?: string;
   CHITTYREGISTER_URL?: string;
   CHITTYCHAT_DATA_API?: string;
   CHITTYSCHEMA_URL?: string;