feat(workspace-studio): Roux Ingest HTTP-mode add-on endpoint

meta/intent.ts

@@ -241,6 +241,60 @@ export async function createIntent(env: IntentEnv, input: CreateIntentInput): Pr
   return rowToIntent(rows[0]);
 }
 
+/**
+ * Atomic create-or-fetch for roux_ingest intents keyed by Gmail message_id.
+ *
+ * Backed by the partial unique index `cc_intents_roux_ingest_message_id_uidx`
+ * (migration 0017). Two concurrent retries of the same Gmail event race on
+ * INSERT; the loser hits the unique violation and we re-SELECT to return the
+ * winner's row. Eliminates the TOCTOU window of SELECT-then-INSERT.
+ *
+ * @canon: chittycanon://gov/governance#classification-axes  STATUS:PENDING
+ */
+export async function createRouxIngestIntentIdempotent(
+  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 = 'roux_ingest'
+        AND payload->'source'->>'message_id' IS NOT NULL
+      DO NOTHING
+    RETURNING *`;
+  if (inserted[0]) {
+    return { intent: rowToIntent(inserted[0]), created: true };
+  }
+  // Conflict — re-fetch the winner.
+  const winner = await sql`
+    SELECT * FROM cc_intents
+    WHERE intent_type = 'roux_ingest'
+      AND payload->'source'->>'message_id' = ${input.messageId}
+    LIMIT 1`;
+  if (!winner[0]) {
+    throw new Error(`ON CONFLICT path with no winning row for 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/0017_roux_ingest_idempotency.sql

@@ -0,0 +1,19 @@
+-- 0017_roux_ingest_idempotency.sql
+--
+-- Partial unique index on Gmail message_id for intent_type='roux_ingest'.
+-- Backs the atomic INSERT ... ON CONFLICT DO NOTHING idempotency guard in
+-- src/routes/workspace-studio.ts. Without this, two concurrent Google retries
+-- of the same Gmail event can both pass a SELECT pre-check before either
+-- INSERT lands, producing duplicate intents and double-fanned-out side
+-- effects.
+--
+-- Partial (WHERE intent_type = 'roux_ingest') so other intent_types that
+-- happen to carry payload->source->>message_id (e.g. future SMS sources) are
+-- not constrained.
+--
+-- @canon: chittycanon://core/services/chittycommand/workspace-studio
+
+CREATE UNIQUE INDEX IF NOT EXISTS cc_intents_roux_ingest_message_id_uidx
+  ON cc_intents ((payload->'source'->>'message_id'))
+  WHERE intent_type = 'roux_ingest'
+    AND payload->'source'->>'message_id' IS NOT NULL;

package.json

@@ -28,6 +28,7 @@
     "drizzle-orm": "^0.45.2",
     "hono": "^4.12.23",
     "hono-agents": "^3.0.7",
+    "jose": "^6.2.3",
     "workers-ai-provider": "^3.1.8",
     "zod": "^4.3.6"
   },

src/index.ts

@@ -34,6 +34,7 @@ import { jobRoutes } from './routes/jobs';
 import { transactionRoutes } from './routes/transactions';
 import { timelineRoutes } from './routes/timeline';
 import { triageRoutes } from './routes/triage';
+import { workspaceStudioRoutes } from './routes/workspace-studio';
 
 // Re-export ActionAgent DO class so the runtime can find it
 export { ActionAgent } from './agents/action-agent';
@@ -70,6 +71,13 @@ export type Env = {
   PLAID_CLIENT_ID?: string;
   PLAID_SECRET?: string;
   PLAID_ENV?: string;
+  // ChittyRoux × Workspace Studio integration
+  // @canon: chittycanon://core/services/chittycommand/workspace-studio
+  CHITTYROUX_GCP_SA_EMAIL?: string;
+  CHITTYROUX_MARKETPLACE_OAUTH_CLIENT_ID?: string;
+  CHITTYROUX_MARKETPLACE_OAUTH_CLIENT_SECRET?: string;
+  REGISTERED_CHANNELS_JSON?: string;
+  GCP_JWKS_URL?: string;
 };
 
 const app = new Hono<{ Bindings: Env; Variables: AuthVariables }>();
@@ -167,6 +175,12 @@ app.route('/api/v1', timelineRoutes);
 app.use('/agent/*', authMiddleware);
 app.use('/agent/*', agentsMiddleware());
 
+// Workspace Studio HTTP-mode add-on endpoints (Roux Ingest custom step).
+// Auth is per-request via authorizationEventObject JWT verification — NOT
+// the global /api/* authMiddleware. Mount outside /api/*.
+// @canon: chittycanon://core/services/chittycommand/workspace-studio
+app.route('/workspace/studio/roux-ingest', workspaceStudioRoutes);
+
 // MCP server — authenticated via shared token in KV
 app.use('/mcp/*', mcpAuthMiddleware);
 app.route('/mcp', mcpRoutes);

src/lib/channel-registry.ts

@@ -0,0 +1,79 @@
+/**
+ * Channel registry resolver.
+ *
+ * Per the global Channel Registration Protocol (CLAUDE.md), every channel/MCP
+ * client that talks to the ChittyOS ecosystem registers via
+ * `agent.chitty.cc/api/v1/channels/register` and receives back a ChittyID and
+ * capability manifest. This module is the chittycommand-side lookup.
+ *
+ * v1 (this file): env-var allowlist parsed from `REGISTERED_CHANNELS_JSON`
+ *                 (a JSON object keyed by channel_id).
+ *
+ * v2 (planned):   HTTP call to `agent.chitty.cc/api/v1/channels/{channelId}`
+ *                 with the same return shape. See TODO below.
+ *
+ * @canon: chittycanon://core/services/chittycommand/channel-registry
+ */
+
+import type { Env } from '../index';
+
+export interface ChannelMeta {
+  channel_id: string;
+  chitty_id: string;
+  platform: string;
+  capabilities: string[];
+  contact_endpoint?: string;
+  status: 'active' | 'suspended' | 'pending';
+}
+
+interface ChannelRegistryEnv extends Pick<Env, 'COMMAND_KV'> {
+  REGISTERED_CHANNELS_JSON?: string;
+}
+
+/**
+ * The canonical Workspace Studio channel ChittyID — used when registering the
+ * Workspace Add-on as a channel under the universal protocol. Surface this
+ * value so it can be added to REGISTERED_CHANNELS_JSON.
+ */
+export const WORKSPACE_STUDIO_CHANNEL_ID = 'chitty:channel:workspace-studio-gmail';
+
+/**
+ * Resolve a channel's metadata, or null if the channel is not registered or
+ * not active.
+ *
+ * TODO(v2): Replace env-var lookup with a fetch to
+ *           `agent.chitty.cc/api/v1/channels/{channelId}` and cache the result
+ *           in `COMMAND_KV` under `channel:meta:{channel_id}` with a 5-minute
+ *           TTL. Same return shape — callers do not need to change.
+ */
+export async function verifyRegisteredChannel(
+  channelId: string,
+  env: ChannelRegistryEnv,
+  requiredCapabilities: string[] = [],
+): Promise<ChannelMeta | null> {
+  if (!channelId) return null;
+
+  const raw = env.REGISTERED_CHANNELS_JSON;
+  if (!raw) return null;
+  let parsed: Record<string, ChannelMeta>;
+  try {
+    parsed = JSON.parse(raw) as Record<string, ChannelMeta>;
+  } catch {
+    console.warn('[channel-registry] REGISTERED_CHANNELS_JSON is not valid JSON');
+    return null;
+  }
+  const meta = parsed[channelId];
+  if (!meta) return null;
+  if (meta.status !== 'active') return null;
+  // Capability gate: the channel must declare every required capability. A
+  // registered+active channel that lacks the capability for THIS operation
+  // (e.g. an SMS channel asked to perform `gmail.ingest`) must be rejected
+  // so the capability manifest is actually enforced.
+  if (requiredCapabilities.length > 0) {
+    const caps = new Set(meta.capabilities ?? []);
+    for (const cap of requiredCapabilities) {
+      if (!caps.has(cap)) return null;
+    }
+  }
+  return meta;
+}

src/lib/dispute-sync.ts

@@ -35,11 +35,45 @@ interface DisputeCore {
 }
 
 // @canon: chittycanon://gov/governance#classification-axes  STATUS:PENDING
+export type RouxPrivilege = 'privileged' | 'pii' | 'hoa_evidentiary' | 'public';
+export type RouxSpace = 'business' | 'legalink';
+
+// Strictness ordering — higher = more sensitive. Used by mergeRouxClassification
+// to take the more sensitive of two derived roux candidates.
+const PRIVILEGE_RANK: Record<RouxPrivilege, number> = {
+  public: 0,
+  hoa_evidentiary: 1,
+  pii: 2,
+  privileged: 3,
+};
+const SPACE_RANK: Record<RouxSpace, number> = {
+  business: 0,
+  legalink: 1,
+};
+
+/**
+ * Pick the MORE sensitive of two Roux classifications. Privilege and space are
+ * compared independently — e.g. (public, legalink) merged with (privileged,
+ * business) yields (privileged, legalink). This prevents an upstream "public"
+ * classification from masking a caller-supplied "legal" dispute_type.
+ *
+ * @canon: chittycanon://gov/governance#classification-axes  STATUS:PENDING
+ */
+export function mergeRouxClassification(
+  a: { privilege: RouxPrivilege; space: RouxSpace },
+  b: { privilege: RouxPrivilege; space: RouxSpace },
+): { privilege: RouxPrivilege; space: RouxSpace } {
+  return {
+    privilege: PRIVILEGE_RANK[a.privilege] >= PRIVILEGE_RANK[b.privilege] ? a.privilege : b.privilege,
+    space: SPACE_RANK[a.space] >= SPACE_RANK[b.space] ? a.space : b.space,
+  };
+}
+
 // Map a cc_disputes.dispute_type to default Roux (privilege, space).
 // Explicit caller-supplied values always override (Q1=(c) pass-through derive).
 export function deriveRouxFromType(disputeType: string): {
-  privilege: 'privileged' | 'pii' | 'hoa_evidentiary' | 'public';
-  space: 'business' | 'legalink';
+  privilege: RouxPrivilege;
+  space: RouxSpace;
 } {
   // Fail-safe routing: dispute_type is free-text from the API/UI. Any string
   // containing "legal" routes to privileged/legalink (prevents leakage of