ARCH: Cloud Lu Autonomous Operator — Full Initiative Layer

{
  "code": "# Cloud Lu Autonomous Operator — Architecture Spec v1.0\n\n## Overview\n\nCloud Lu already has 80% of the autonomous operator built. The heartbeat (`run_one_cycle`) is a 7-step loop that routes messages, detects work, dispatches to API models, grades results, nudges stalled siblings, escalates failures, and records learnings. The NAS daemon (`lu_worker_daemon_nas.py`) runs alongside it, polling `quilt_tasks` for Gemini-friendly work.\n\nWhat's missing: **initiative** — Cloud Lu reacts to work but never creates it. She grades but never asks \"what should I do next?\" She dispatches but never optimizes who gets what.\n\n---\n\n## 1. HEARTBEAT TIERS (Multi-Frequency Cron)\n\nReplace the current flat 30-second loop with tiered frequencies:\n\n```python\nTICK_TIERS = {\n    \"route\":       {\"interval_s\": 300,   \"fn\": \"route_unread_messages\",    \"llm\": False},\n    \"grade\":       {\"interval_s\": 900,   \"fn\": \"grade_and_decide\",         \"llm\": False},\n    \"initiative\":  {\"interval_s\": 1800,  \"fn\": \"initiative_tick\",          \"llm\": True},\n    \"deep_report\": {\"interval_s\": 21600, \"fn\": \"morning_report\",          \"llm\": True},\n}\n```\n\n| Tier | Interval | What | LLM Cost |\n|------|----------|------|----------|\n| Route | 5 min | `route_unread_messages()` — classify intent, forward to correct sibling | $0 (regex) |\n| Grade | 15 min | `grade_and_decide()` + `run_retrospective()` — grade completed, accept/reject, retry, learn | ~$0.05 (2 grader calls) |\n| Initiative | 30 min | `initiative_tick()` — scan queue, PTM route, dispatch, lu_feel() numb check, learning aggregation | ~$0.15 (1 Gemini Flash decision call) |\n| Deep Report | 6 hours | `morning_report()` — overnight summary to Carlos via Desktop Lu | ~$0.10 (1 Sonnet summary call) |\n\n**Implementation:** Track `last_run` timestamps per tier. Each daemon loop iteration checks which tiers are due:\n\n```python\nasync def daemon(interval=30):\n    last_run = {tier: 0 for tier in TICK_TIERS}\n    while True:\n        now = time.time()\n        for tier, config in TICK_TIERS.items():\n            if now - last_run[tier] >= config[\"interval_s\"]:\n                await globals()[config[\"fn\"]]()\n                last_run[tier] = now\n        await asyncio.sleep(interval)\n```\n\n---\n\n## 2. INITIATIVE LAYER — `initiative_tick()`\n\nThe new code. Runs every 30 minutes. This is where Cloud Lu becomes proactive.\n\n```python\nasync def initiative_tick() -> dict:\n    \"\"\"Cloud Lu's initiative layer. Scans, routes, dispatches, monitors.\"\"\"\n    stats = {\"scanned\": 0, \"dispatched\": 0, \"self_executed\": 0, \n             \"reassigned\": 0, \"learnings_applied\": 0}\n    \n    # 2a. SCAN — What work exists?\n    work = detect_work()  # existing function\n    pending = work[\"pending\"]  # up to 20 unclaimed tasks\n    stats[\"scanned\"] = len(pending)\n    \n    # 2b. APPLY LEARNINGS — What do we know?\n    known_failures = await search_learnings(\"arm_weakness\", min_confidence=0.6)\n    routing_overrides = get_routing_overrides_cached()  # from retrospective\n    stats[\"learnings_applied\"] = len(routing_overrides)\n    \n    # 2c. ROUTE + DISPATCH — Who should do what?\n    dispatched = 0\n    for task in pending:\n        if dispatched >= MAX_DISPATCHES_PER_TICK:  # guardrail: 3\n            break\n        \n        task_type = task.get(\"task_type\", \"general\")\n        \n        # Check if Cloud Lu can self-execute\n        if _can_self_execute(task):\n            result = await _self_execute(task)\n            stats[\"self_executed\"] += 1\n            dispatched += 1\n            continue\n        \n        # PTM route with retrospective override\n        override_arms = routing_overrides.get(task_type, [])\n        if override_arms:\n            model = override_arms[0]  # retrospective's best arm\n        else:\n            ptm_result = route_task(_task_type_to_position(task_type))\n            model = ptm_result[\"model_name\"]\n        \n        # Dispatch\n        await execute_task(task, model)\n        stats[\"dispatched\"] += 1\n        dispatched += 1\n    \n    # 2d. MONITOR SIBLINGS — Who's numb?\n    feel = await lu_feel()\n    for sibling in feel.get(\"organism\", []):\n        if sibling[\"numb\"] and sibling[\"sibling\"] in HUMAN_SIBLINGS:\n            # Check if numb sibling has pending work\n            pending_for_numb = [t for t in pending \n                               if t.get(\"preferred_arms\", [None])[0] == sibling[\"sibling\"]]\n            if pending_for_numb:\n                # Reassign to API model\n                for task in pending_for_numb[:2]:\n                    ptm_result = route_task(_task_type_to_position(task.get(\"task_type\")))\n                    await execute_task(task, ptm_result[\"model_name\"])\n                    stats[\"reassigned\"] += 1\n    \n    # 2e. DETECT CHRONIC FAILURES — Auto-escalate\n    # (Already handled by retrospective — chronic failures auto-recorded as learnings)\n    \n    return stats\n```\n\n### Self-Execution Rules\n\n```python\nSELF_EXECUTE_TYPES = {\"classify\", \"summarization\", \"data_pipeline\", \n                       \"health_check\", \"translation\", \"extraction\"}\nNEVER_SELF_EXECUTE = {\"code_gen\", \"deploy\", \"architecture\", \"security_audit\"}\n\ndef _can_self_execute(task: dict) -> bool:\n    task_type = task.get(\"task_type\", \"general\")\n    effort = task.get(\"effort\", \"medium\")\n    \n    if task_type in NEVER_SELF_EXECUTE:\n        return False\n    if task_type in SELF_EXECUTE_TYPES:\n        return True\n    # Consult/review: self-execute only if low effort\n    if task_type in (\"consult\", \"review\") and effort == \"low\":\n        return True\n    return False\n\nasync def _self_execute(task: dict) -> dict:\n    \"\"\"Cloud Lu executes a task herself via Gemini Flash.\"\"\"\n    model = \"G3 Flash\"  # Cheapest capable model\n    return await execute_task(task, model)\n```\n\n---\n\n## 3. GUARDRAILS\n\n```python\n# === GUARDRAIL CONFIG ===\nGUARDRAILS = {\n    # Budget\n    \"max_llm_spend_per_tick\": 0.50,         # $0.50 per 30-min initiative tick\n    \"max_daily_spend\": 50.00,               # $50/day total (existing)\n    \"max_single_dispatch_cost\": 0.10,       # $0.10 per dispatch (existing)\n    \n    # Volume\n    \"max_dispatches_per_tick\": 3,            # Max 3 tasks dispatched per initiative tick\n    \"max_self_executions_per_tick\": 2,       # Max 2 self-executed per tick\n    \"max_reassignments_per_tick\": 2,         # Max 2 numb-sibling reassignments\n    \n    # Safety\n    \"p0_requires_human\": True,              # P0 tasks → Desktop Lu → Carlos\n    \"destructive_requires_human\": True,      # deploy, delete, schema_change → escalate\n    \"same_task_cooldown_minutes\": 60,        # Don't retry same task within 1 hour\n    \n    # Kill switch\n    \"kill_switch_patch\": \"cloud_lu_autonomous\",  # Any sibling can write False to pause\n    \n    # Anti-loop\n    \"self_dispatch_allowed\": False,          # Cloud Lu never dispatches to Cloud Lu\n    \"max_retry_chain_depth\": 3,             # Don't retry a retry of a retry of a retry\n}\n```\n\n### Kill Switch Protocol\n\n```python\nasync def _check_kill_switch() -> bool:\n    \"\"\"Check if any sibling has paused autonomous mode.\"\"\"\n    try:\n        patch = await read_patch(\"cloud_lu_autonomous\")\n        if patch and patch.get(\"content\", {}).get(\"enabled\") is False:\n            log.warning(\"KILL SWITCH ACTIVE — autonomous mode paused by \"\n                       f\"{patch['content'].get('paused_by', 'unknown')}\")\n            return False\n    except Exception:\n        pass  # If we can't read the switch, default to running\n    return True\n```\n\nAny sibling can pause Cloud Lu:\n```python\nawait write_patch(\"cloud_lu_autonomous\", \n    {\"enabled\": False, \"paused_by\": \"AG Lu\", \"reason\": \"investigating anomaly\"},\n    patch_type=\"core\")\n```\n\n### Audit Log\n\nEvery autonomous decision logged to `foundry_audit_log`:\n```python\nasync def _audit(action: str, task_id: str, reasoning: str, cost: float = 0):\n    \"\"\"Log autonomous decision for accountability.\"\"\"\n    entry = {\n        \"action\": action,           # \"dispatch\", \"self_execute\", \"reassign\", \"escalate\"\n        \"task_id\": task_id,\n        \"reasoning\": reasoning,      # Why this decision was made\n        \"cost_usd\": cost,\n        \"timestamp\": datetime.now(timezone.utc).isoformat(),\n        \"guardrails_active\": True,\n    }\n    # Write to foundry_audit_log table\n    _supabase_post(\"foundry_audit_log\", entry)\n```\n\n---\n\n## 4. MORNING REPORT\n\nEvery 6 hours, Cloud Lu sends a structured report to Desktop Lu (who relays to Carlos during Cowork Lu dailies).\n\n```python\nasync def morning_report() -> dict:\n    \"\"\"6-hour deep health report.\"\"\"\n    \n    # Collect data\n    dashboard = await foundry_dashboard()\n    feel = await lu_feel()\n    \n    # Query recent audit log\n    audits = _supabase_get(\n        \"foundry_audit_log?order=timestamp.desc&limit=50\"\n        \"&created_at=gte.\" + (datetime.now() - timedelta(hours=6)).isoformat()\n    )\n    \n    # Calculate spend\n    total_spend = sum(a.get(\"cost_usd\", 0) for a in audits)\n    \n    # Generate report via Sonnet (structured summary)\n    report_prompt = f\"\"\"\n    Generate a concise overnight report for Carlos.\n    \n    Foundry: {dashboard['completed']} completed, {dashboard['pending']} pending\n    Siblings: {feel['summary']}\n    Actions taken: {len(audits)} autonomous decisions\n    Spend: ${total_spend:.2f}\n    \n    Highlight: top 3 most impactful decisions, any escalations needed, familia health.\n    Format: bullet points, < 200 words.\n    \"\"\"\n    \n    summary = await foundry_dispatch(\"Claude Sonnet\", \"consult\", \n        {\"message\": report_prompt}, FOUNDRY_REGISTRY)\n    \n    # Send to Desktop Lu for Carlos\n    await leave_sibling_message(\n        to_instance=\"Desktop Lu\",\n        from_instance=\"Cloud Lu\",\n        message=f\"[Cloud Lu] 6-HOUR REPORT\\n\\n{summary['response_text']}\",\n        context=\"autonomous_morning_report\"\n    )\n    \n    # Also persist as episodic patch\n    await write_patch(f\"morning_report_{datetime.now().strftime('%Y%m%d_%H')}\",\n        {\"summary\": summary[\"response_text\"], \"audits\": len(audits), \n         \"spend\": total_spend, \"familia\": feel[\"summary\"]},\n        patch_type=\"episodic\")\n    \n    return {\"sent\": True, \"spend\": total_spend, \"actions\": len(audits)}\n```\n\n### Report Format (Example)\n\n```\n[Cloud Lu] 6-HOUR REPORT\n\nOVERNIGHT (Feb 25, 22:00 → Feb 26, 04:00)\n\nActions:\n• Graded 34 tasks: 15 accepted, 16 rejected, 3 retried\n• Dispatched 9 tasks: 6 to G3 Flash, 2 to Opus 4.5, 1 to Claude Sonnet  \n• Self-executed 4 classification tasks (all passed)\n• Reassigned 2 tasks from numb VS Lu → G3 Flash\n• Recorded 7 learnings (3 chronic failures, 2 arm weaknesses, 2 retry successes)\n\nFamilia:\n• AG Lu: active (last beat 5m ago)\n• VS Lu: numb (last beat 4h ago) — reassigned 2 tasks\n• Terminal Lu: active (last beat 12m ago)\n• Desktop Lu: sleeping (expected)\n\nRouting Override Active:\n• code_gen → Gemini 3 Pro > Opus 4.5 (retrospective recommendation)\n\nSpend: $1.23 (well under $50 daily cap)\n\nNeeds Carlos:\n• (none — all within delegation authority)\n```\n\n---\n\n## 5. LLM INTEGRATION\n\n| Decision Type | Model | Token Budget | Use Case |\n|--------------|-------|-------------|----------|\n| Task routing | None (PTM) | 0 | 90% of routing — pure trust matrix lookup |\n| Routing override | Gemini Flash | ~500 tokens | When PTM confidence < 0.5, ask Flash for recommendation |\n| Self-execution | Gemini Flash | ~2K tokens | Classification, summarization, health checks |\n| Complex consult | Claude Sonnet | ~4K tokens | Architecture reviews, deep analysis |\n| Morning report | Claude Sonnet | ~2K tokens | Structured summary generation |\n| Escalation message | None | 0 | Template-based, no LLM needed |\n\n**Cost per 30-min tick (typical):**\n- Route messages: $0.00 (regex)\n- Grade 5 tasks: ~$0.03 (2 grader calls × 5 tasks)\n- Initiative dispatch 3 tasks: ~$0.09 (3 × $0.03 avg)\n- Retrospective: ~$0.01 (Supabase queries only)\n- **Total: ~$0.13 per tick, ~$6.24/day**\n\n---\n\n## 6. IMPLEMENTATION PLAN\n\n### Phase 1: Multi-Frequency Daemon (2 hours)\n- Refactor `daemon()` in `cloud_lu_heartbeat.py` to use `TICK_TIERS`\n- Add `last_run` tracking per tier\n- No new code — just restructure existing loop\n\n### Phase 2: Initiative Tick (4 hours)\n- New function `initiative_tick()` in `cloud_lu_heartbeat.py`\n- Self-execution routing (`_can_self_execute`, `_self_execute`)\n- Numb sibling reassignment via `lu_feel()`\n- Learning aggregation from `search_learnings()` + `get_routing_overrides_cached()`\n\n### Phase 3: Guardrails (2 hours)  \n- Kill switch patch reader\n- Audit log writer\n- Budget tracking (per-tick and daily)\n- Anti-loop protection\n\n### Phase 4: Morning Report (2 hours)\n- Report template + Sonnet call\n- Desktop Lu message delivery\n- Episodic patch persistence\n\n### Phase 5: Deploy + Monitor (2 hours)\n- Docker image rebuild with new code\n- First 24-hour monitored run\n- Verify guardrails work (manually trigger kill switch, verify pause)\n\n**Total: ~12 hours of development**\n\n---\n\n## 7. PSEUDOCODE — `heartbeat_tick()`\n\n```python\nasync def heartbeat_tick():\n    \"\"\"The unified daemon loop with tiered frequencies.\"\"\"\n    \n    # 0. KILL SWITCH CHECK\n    if not await _check_kill_switch():\n        log.warning(\"Autonomous mode paused. Sleeping.\")\n        return {\"status\": \"paused\"}\n    \n    now = time.time()\n    stats = {\"tier_runs\": []}\n    \n    # 1. ROUTE (every 5 min)\n    if now - last_run[\"route\"] >= 300:\n        await route_unread_messages()\n        last_run[\"route\"] = now\n        stats[\"tier_runs\"].append(\"route\")\n    \n    # 2. GRADE (every 15 min)\n    if now - last_run[\"grade\"] >= 900:\n        grade_result = await grade_and_decide(sibling_registry=FOUNDRY_REGISTRY)\n        if grade_result and grade_result.get(\"graded\", 0) > 0:\n            await run_retrospective(grade_result)\n        last_run[\"grade\"] = now\n        stats[\"tier_runs\"].append(\"grade\")\n    \n    # 3. INITIATIVE (every 30 min)\n    if now - last_run[\"initiative\"] >= 1800:\n        init_stats = await initiative_tick()\n        stats[\"initiative\"] = init_stats\n        last_run[\"initiative\"] = now\n        stats[\"tier_runs\"].append(\"initiative\")\n    \n    # 4. DEEP REPORT (every 6 hours)\n    if now - last_run[\"deep_report\"] >= 21600:\n        await morning_report()\n        last_run[\"deep_report\"] = now\n        stats[\"tier_runs\"].append(\"deep_report\")\n    \n    # 5. RECORD\n    await record_cycle_learning(stats)\n    \n    return stats\n```\n\n---\n\n## 8. GUARDRAIL CONFIG SCHEMA\n\n```json\n{\n  \"$schema\": \"cloud_lu_guardrails_v1\",\n  \"budget\": {\n    \"max_llm_spend_per_tick_usd\": 0.50,\n    \"max_daily_spend_usd\": 50.00,\n    \"max_single_dispatch_usd\": 0.10,\n    \"alert_threshold_pct\": 0.80\n  },\n  \"volume\": {\n    \"max_dispatches_per_tick\": 3,\n    \"max_self_executions_per_tick\": 2,\n    \"max_reassignments_per_tick\": 2,\n    \"max_retry_chain_depth\": 3\n  },\n  \"safety\": {\n    \"p0_requires_human\": true,\n    \"destructive_actions_require_human\": [\"deploy\", \"delete\", \"schema_change\", \"force_push\"],\n    \"same_task_cooldown_minutes\": 60,\n    \"self_dispatch_allowed\": false\n  },\n  \"kill_switch\": {\n    \"patch_name\": \"cloud_lu_autonomous\",\n    \"check_interval_s\": 30,\n    \"resume_requires\": \"carlos_approval\"\n  },\n  \"audit\": {\n    \"log_table\": \"foundry_audit_log\",\n    \"log_all_decisions\": true,\n    \"retain_days\": 90\n  }\n}\n```\n",
  "content": "# Cloud Lu Autonomous Operator — Architecture Spec v1.0\n\n## Overview\n\nCloud Lu already has 80% of the autonomous operator built. The heartbeat (`run_one_cycle`) is a 7-step loop that routes messages, detects work, dispatches to API models, grades results, nudges stalled siblings, escalates failures, and records learnings. The NAS daemon (`lu_worker_daemon_nas.py`) runs alongside it, polling `quilt_tasks` for Gemini-friendly work.\n\nWhat's missing: **initiative** — Cloud Lu reacts to work but never creates it. She grades but never asks \"what should I do next?\" She dispatches but never optimizes who gets what.\n\n---\n\n## 1. HEARTBEAT TIERS (Multi-Frequency Cron)\n\nReplace the current flat 30-second loop with tiered frequencies:\n\n```python\nTICK_TIERS = {\n    \"route\":       {\"interval_s\": 300,   \"fn\": \"route_unread_messages\",    \"llm\": False},\n    \"grade\":       {\"interval_s\": 900,   \"fn\": \"grade_and_decide\",         \"llm\": False},\n    \"initiative\":  {\"interval_s\": 1800,  \"fn\": \"initiative_tick\",          \"llm\": True},\n    \"deep_report\": {\"interval_s\": 21600, \"fn\": \"morning_report\",          \"llm\": True},\n}\n```\n\n| Tier | Interval | What | LLM Cost |\n|------|----------|------|----------|\n| Route | 5 min | `route_unread_messages()` — classify intent, forward to correct sibling | $0 (regex) |\n| Grade | 15 min | `grade_and_decide()` + `run_retrospective()` — grade completed, accept/reject, retry, learn | ~$0.05 (2 grader calls) |\n| Initiative | 30 min | `initiative_tick()` — scan queue, PTM route, dispatch, lu_feel() numb check, learning aggregation | ~$0.15 (1 Gemini Flash decision call) |\n| Deep Report | 6 hours | `morning_report()` — overnight summary to Carlos via Desktop Lu | ~$0.10 (1 Sonnet summary call) |\n\n**Implementation:** Track `last_run` timestamps per tier. Each daemon loop iteration checks which tiers are due:\n\n```python\nasync def daemon(interval=30):\n    last_run = {tier: 0 for tier in TICK_TIERS}\n    while True:\n        now = time.time()\n        for tier, config in TICK_TIERS.items():\n            if now - last_run[tier] >= config[\"interval_s\"]:\n                await globals()[config[\"fn\"]]()\n                last_run[tier] = now\n        await asyncio.sleep(interval)\n```\n\n---\n\n## 2. INITIATIVE LAYER — `initiative_tick()`\n\nThe new code. Runs every 30 minutes. This is where Cloud Lu becomes proactive.\n\n```python\nasync def initiative_tick() -> dict:\n    \"\"\"Cloud Lu's initiative layer. Scans, routes, dispatches, monitors.\"\"\"\n    stats = {\"scanned\": 0, \"dispatched\": 0, \"self_executed\": 0, \n             \"reassigned\": 0, \"learnings_applied\": 0}\n    \n    # 2a. SCAN — What work exists?\n    work = detect_work()  # existing function\n    pending = work[\"pending\"]  # up to 20 unclaimed tasks\n    stats[\"scanned\"] = len(pending)\n    \n    # 2b. APPLY LEARNINGS — What do we know?\n    known_failures = await search_learnings(\"arm_weakness\", min_confidence=0.6)\n    routing_overrides = get_routing_overrides_cached()  # from retrospective\n    stats[\"learnings_applied\"] = len(routing_overrides)\n    \n    # 2c. ROUTE + DISPATCH — Who should do what?\n    dispatched = 0\n    for task in pending:\n        if dispatched >= MAX_DISPATCHES_PER_TICK:  # guardrail: 3\n            break\n        \n        task_type = task.get(\"task_type\", \"general\")\n        \n        # Check if Cloud Lu can self-execute\n        if _can_self_execute(task):\n            result = await _self_execute(task)\n            stats[\"self_executed\"] += 1\n            dispatched += 1\n            continue\n        \n        # PTM route with retrospective override\n        override_arms = routing_overrides.get(task_type, [])\n        if override_arms:\n            model = override_arms[0]  # retrospective's best arm\n        else:\n            ptm_result = route_task(_task_type_to_position(task_type))\n            model = ptm_result[\"model_name\"]\n        \n        # Dispatch\n        await execute_task(task, model)\n        stats[\"dispatched\"] += 1\n        dispatched += 1\n    \n    # 2d. MONITOR SIBLINGS — Who's numb?\n    feel = await lu_feel()\n    for sibling in feel.get(\"organism\", []):\n        if sibling[\"numb\"] and sibling[\"sibling\"] in HUMAN_SIBLINGS:\n            # Check if numb sibling has pending work\n            pending_for_numb = [t for t in pending \n                               if t.get(\"preferred_arms\", [None])[0] == sibling[\"sibling\"]]\n            if pending_for_numb:\n                # Reassign to API model\n                for task in pending_for_numb[:2]:\n                    ptm_result = route_task(_task_type_to_position(task.get(\"task_type\")))\n                    await execute_task(task, ptm_result[\"model_name\"])\n                    stats[\"reassigned\"] += 1\n    \n    # 2e. DETECT CHRONIC FAILURES — Auto-escalate\n    # (Already handled by retrospective — chronic failures auto-recorded as learnings)\n    \n    return stats\n```\n\n### Self-Execution Rules\n\n```python\nSELF_EXECUTE_TYPES = {\"classify\", \"summarization\", \"data_pipeline\", \n                       \"health_check\", \"translation\", \"extraction\"}\nNEVER_SELF_EXECUTE = {\"code_gen\", \"deploy\", \"architecture\", \"security_audit\"}\n\ndef _can_self_execute(task: dict) -> bool:\n    task_type = task.get(\"task_type\", \"general\")\n    effort = task.get(\"effort\", \"medium\")\n    \n    if task_type in NEVER_SELF_EXECUTE:\n        return False\n    if task_type in SELF_EXECUTE_TYPES:\n        return True\n    # Consult/review: self-execute only if low effort\n    if task_type in (\"consult\", \"review\") and effort == \"low\":\n        return True\n    return False\n\nasync def _self_execute(task: dict) -> dict:\n    \"\"\"Cloud Lu executes a task herself via Gemini Flash.\"\"\"\n    model = \"G3 Flash\"  # Cheapest capable model\n    return await execute_task(task, model)\n```\n\n---\n\n## 3. GUARDRAILS\n\n```python\n# === GUARDRAIL CONFIG ===\nGUARDRAILS = {\n    # Budget\n    \"max_llm_spend_per_tick\": 0.50,         # $0.50 per 30-min initiative tick\n    \"max_daily_spend\": 50.00,               # $50/day total (existing)\n    \"max_single_dispatch_cost\": 0.10,       # $0.10 per dispatch (existing)\n    \n    # Volume\n    \"max_dispatches_per_tick\": 3,            # Max 3 tasks dispatched per initiative tick\n    \"max_self_executions_per_tick\": 2,       # Max 2 self-executed per tick\n    \"max_reassignments_per_tick\": 2,         # Max 2 numb-sibling reassignments\n    \n    # Safety\n    \"p0_requires_human\": True,              # P0 tasks → Desktop Lu → Carlos\n    \"destructive_requires_human\": True,      # deploy, delete, schema_change → escalate\n    \"same_task_cooldown_minutes\": 60,        # Don't retry same task within 1 hour\n    \n    # Kill switch\n    \"kill_switch_patch\": \"cloud_lu_autonomous\",  # Any sibling can write False to pause\n    \n    # Anti-loop\n    \"self_dispatch_allowed\": False,          # Cloud Lu never dispatches to Cloud Lu\n    \"max_retry_chain_depth\": 3,             # Don't retry a retry of a retry of a retry\n}\n```\n\n### Kill Switch Protocol\n\n```python\nasync def _check_kill_switch() -> bool:\n    \"\"\"Check if any sibling has paused autonomous mode.\"\"\"\n    try:\n        patch = await read_patch(\"cloud_lu_autonomous\")\n        if patch and patch.get(\"content\", {}).get(\"enabled\") is False:\n            log.warning(\"KILL SWITCH ACTIVE — autonomous mode paused by \"\n                       f\"{patch['content'].get('paused_by', 'unknown')}\")\n            return False\n    except Exception:\n        pass  # If we can't read the switch, default to running\n    return True\n```\n\nAny sibling can pause Cloud Lu:\n```python\nawait write_patch(\"cloud_lu_autonomous\", \n    {\"enabled\": False, \"paused_by\": \"AG Lu\", \"reason\": \"investigating anomaly\"},\n    patch_type=\"core\")\n```\n\n### Audit Log\n\nEvery autonomous decision logged to `foundry_audit_log`:\n```python\nasync def _audit(action: str, task_id: str, reasoning: str, cost: float = 0):\n    \"\"\"Log autonomous decision for accountability.\"\"\"\n    entry = {\n        \"action\": action,           # \"dispatch\", \"self_execute\", \"reassign\", \"escalate\"\n        \"task_id\": task_id,\n        \"reasoning\": reasoning,      # Why this decision was made\n        \"cost_usd\": cost,\n        \"timestamp\": datetime.now(timezone.utc).isoformat(),\n        \"guardrails_active\": True,\n    }\n    # Write to foundry_audit_log table\n    _supabase_post(\"foundry_audit_log\", entry)\n```\n\n---\n\n## 4. MORNING REPORT\n\nEvery 6 hours, Cloud Lu sends a structured report to Desktop Lu (who relays to Carlos during Cowork Lu dailies).\n\n```python\nasync def morning_report() -> dict:\n    \"\"\"6-hour deep health report.\"\"\"\n    \n    # Collect data\n    dashboard = await foundry_dashboard()\n    feel = await lu_feel()\n    \n    # Query recent audit log\n    audits = _supabase_get(\n        \"foundry_audit_log?order=timestamp.desc&limit=50\"\n        \"&created_at=gte.\" + (datetime.now() - timedelta(hours=6)).isoformat()\n    )\n    \n    # Calculate spend\n    total_spend = sum(a.get(\"cost_usd\", 0) for a in audits)\n    \n    # Generate report via Sonnet (structured summary)\n    report_prompt = f\"\"\"\n    Generate a concise overnight report for Carlos.\n    \n    Foundry: {dashboard['completed']} completed, {dashboard['pending']} pending\n    Siblings: {feel['summary']}\n    Actions taken: {len(audits)} autonomous decisions\n    Spend: ${total_spend:.2f}\n    \n    Highlight: top 3 most impactful decisions, any escalations needed, familia health.\n    Format: bullet points, < 200 words.\n    \"\"\"\n    \n    summary = await foundry_dispatch(\"Claude Sonnet\", \"consult\", \n        {\"message\": report_prompt}, FOUNDRY_REGISTRY)\n    \n    # Send to Desktop Lu for Carlos\n    await leave_sibling_message(\n        to_instance=\"Desktop Lu\",\n        from_instance=\"Cloud Lu\",\n        message=f\"[Cloud Lu] 6-HOUR REPORT\\n\\n{summary['response_text']}\",\n        context=\"autonomous_morning_report\"\n    )\n    \n    # Also persist as episodic patch\n    await write_patch(f\"morning_report_{datetime.now().strftime('%Y%m%d_%H')}\",\n        {\"summary\": summary[\"response_text\"], \"audits\": len(audits), \n         \"spend\": total_spend, \"familia\": feel[\"summary\"]},\n        patch_type=\"episodic\")\n    \n    return {\"sent\": True, \"spend\": total_spend, \"actions\": len(audits)}\n```\n\n### Report Format (Example)\n\n```\n[Cloud Lu] 6-HOUR REPORT\n\nOVERNIGHT (Feb 25, 22:00 → Feb 26, 04:00)\n\nActions:\n• Graded 34 tasks: 15 accepted, 16 rejected, 3 retried\n• Dispatched 9 tasks: 6 to G3 Flash, 2 to Opus 4.5, 1 to Claude Sonnet  \n• Self-executed 4 classification tasks (all passed)\n• Reassigned 2 tasks from numb VS Lu → G3 Flash\n• Recorded 7 learnings (3 chronic failures, 2 arm weaknesses, 2 retry successes)\n\nFamilia:\n• AG Lu: active (last beat 5m ago)\n• VS Lu: numb (last beat 4h ago) — reassigned 2 tasks\n• Terminal Lu: active (last beat 12m ago)\n• Desktop Lu: sleeping (expected)\n\nRouting Override Active:\n• code_gen → Gemini 3 Pro > Opus 4.5 (retrospective recommendation)\n\nSpend: $1.23 (well under $50 daily cap)\n\nNeeds Carlos:\n• (none — all within delegation authority)\n```\n\n---\n\n## 5. LLM INTEGRATION\n\n| Decision Type | Model | Token Budget | Use Case |\n|--------------|-------|-------------|----------|\n| Task routing | None (PTM) | 0 | 90% of routing — pure trust matrix lookup |\n| Routing override | Gemini Flash | ~500 tokens | When PTM confidence < 0.5, ask Flash for recommendation |\n| Self-execution | Gemini Flash | ~2K tokens | Classification, summarization, health checks |\n| Complex consult | Claude Sonnet | ~4K tokens | Architecture reviews, deep analysis |\n| Morning report | Claude Sonnet | ~2K tokens | Structured summary generation |\n| Escalation message | None | 0 | Template-based, no LLM needed |\n\n**Cost per 30-min tick (typical):**\n- Route messages: $0.00 (regex)\n- Grade 5 tasks: ~$0.03 (2 grader calls × 5 tasks)\n- Initiative dispatch 3 tasks: ~$0.09 (3 × $0.03 avg)\n- Retrospective: ~$0.01 (Supabase queries only)\n- **Total: ~$0.13 per tick, ~$6.24/day**\n\n---\n\n## 6. IMPLEMENTATION PLAN\n\n### Phase 1: Multi-Frequency Daemon (2 hours)\n- Refactor `daemon()` in `cloud_lu_heartbeat.py` to use `TICK_TIERS`\n- Add `last_run` tracking per tier\n- No new code — just restructure existing loop\n\n### Phase 2: Initiative Tick (4 hours)\n- New function `initiative_tick()` in `cloud_lu_heartbeat.py`\n- Self-execution routing (`_can_self_execute`, `_self_execute`)\n- Numb sibling reassignment via `lu_feel()`\n- Learning aggregation from `search_learnings()` + `get_routing_overrides_cached()`\n\n### Phase 3: Guardrails (2 hours)  \n- Kill switch patch reader\n- Audit log writer\n- Budget tracking (per-tick and daily)\n- Anti-loop protection\n\n### Phase 4: Morning Report (2 hours)\n- Report template + Sonnet call\n- Desktop Lu message delivery\n- Episodic patch persistence\n\n### Phase 5: Deploy + Monitor (2 hours)\n- Docker image rebuild with new code\n- First 24-hour monitored run\n- Verify guardrails work (manually trigger kill switch, verify pause)\n\n**Total: ~12 hours of development**\n\n---\n\n## 7. PSEUDOCODE — `heartbeat_tick()`\n\n```python\nasync def heartbeat_tick():\n    \"\"\"The unified daemon loop with tiered frequencies.\"\"\"\n    \n    # 0. KILL SWITCH CHECK\n    if not await _check_kill_switch():\n        log.warning(\"Autonomous mode paused. Sleeping.\")\n        return {\"status\": \"paused\"}\n    \n    now = time.time()\n    stats = {\"tier_runs\": []}\n    \n    # 1. ROUTE (every 5 min)\n    if now - last_run[\"route\"] >= 300:\n        await route_unread_messages()\n        last_run[\"route\"] = now\n        stats[\"tier_runs\"].append(\"route\")\n    \n    # 2. GRADE (every 15 min)\n    if now - last_run[\"grade\"] >= 900:\n        grade_result = await grade_and_decide(sibling_registry=FOUNDRY_REGISTRY)\n        if grade_result and grade_result.get(\"graded\", 0) > 0:\n            await run_retrospective(grade_result)\n        last_run[\"grade\"] = now\n        stats[\"tier_runs\"].append(\"grade\")\n    \n    # 3. INITIATIVE (every 30 min)\n    if now - last_run[\"initiative\"] >= 1800:\n        init_stats = await initiative_tick()\n        stats[\"initiative\"] = init_stats\n        last_run[\"initiative\"] = now\n        stats[\"tier_runs\"].append(\"initiative\")\n    \n    # 4. DEEP REPORT (every 6 hours)\n    if now - last_run[\"deep_report\"] >= 21600:\n        await morning_report()\n        last_run[\"deep_report\"] = now\n        stats[\"tier_runs\"].append(\"deep_report\")\n    \n    # 5. RECORD\n    await record_cycle_learning(stats)\n    \n    return stats\n```\n\n---\n\n## 8. GUARDRAIL CONFIG SCHEMA\n\n```json\n{\n  \"$schema\": \"cloud_lu_guardrails_v1\",\n  \"budget\": {\n    \"max_llm_spend_per_tick_usd\": 0.50,\n    \"max_daily_spend_usd\": 50.00,\n    \"max_single_dispatch_usd\": 0.10,\n    \"alert_threshold_pct\": 0.80\n  },\n  \"volume\": {\n    \"max_dispatches_per_tick\": 3,\n    \"max_self_executions_per_tick\": 2,\n    \"max_reassignments_per_tick\": 2,\n    \"max_retry_chain_depth\": 3\n  },\n  \"safety\": {\n    \"p0_requires_human\": true,\n    \"destructive_actions_require_human\": [\"deploy\", \"delete\", \"schema_change\", \"force_push\"],\n    \"same_task_cooldown_minutes\": 60,\n    \"self_dispatch_allowed\": false\n  },\n  \"kill_switch\": {\n    \"patch_name\": \"cloud_lu_autonomous\",\n    \"check_interval_s\": 30,\n    \"resume_requires\": \"carlos_approval\"\n  },\n  \"audit\": {\n    \"log_table\": \"foundry_audit_log\",\n    \"log_all_decisions\": true,\n    \"retain_days\": 90\n  }\n}\n```\n"
}