feat: unify daily operating loop under /daily with hardened control-plane receipts

docs/plans/TASK-DAILY-UNIFICATION-spec.md

@@ -0,0 +1,93 @@
+# TASK_SPEC.md: Daily Loop Unification & Agent Control Plane Receipts (Production v5 Final Hardened)
+
+## 1. The Job
+Unify morning operations under **`/daily`** (`--scan` vs `--interactive`) and cleanly remove `/daily-brief`.
+- **Master Entry Point (`plugins/portfolio-advisor/scripts/run_daily.py`)**:
+  - Authoritative entry point lives exclusively in `plugins/portfolio-advisor/scripts/run_daily.py` (NO symlinks into `investment_screener/backend/py_services/`, keeping operational and web application domains strictly decoupled).
+  - Mode-aware execution:
+    - `--scan`: Non-interactive fast morning brief. Executes Step 0 (Readiness) ➔ Step 1 (Morning Brief) ➔ Finalizes (`required_steps = [0, 1]`).
+    - `--interactive` (Default): Full 6-step loop. Executes Step 0 ➔ Step 1 ➔ Step 2 ➔ Step 3 ➔ Step 4 ➔ Step 5 ➔ Finalizes (`required_steps = [0, 1, 2, 3, 4, 5]`).
+- **Strict Run Identity Ownership**:
+  - `run_id` is strictly generated by the runner/control-plane (`DAILY-YYYYMMDD-HHMMSS-<HEX8>`). Callers CANNOT supply an arbitrary `run_id`.
+  - External task correlation is supported solely via an optional `--correlation-id <ID>` metadata field.
+- **Signal Trapping**:
+  - Traps `SIGINT`, `SIGTERM`, and unhandled exceptions to record an auditable terminal `ABORTED` / `FAILED` receipt.
+- **6-Step Finite State Machine (Steps 0–5)**:
+  - **Step 0 — Readiness**: Validates OS substrate viability (`control_plane.db` write transaction with `BEGIN IMMEDIATE`), server status, and domain DB freshness. (Git hook validation is removed from daily investment runs to prevent coupling portfolio operations to dev tooling). Step 0 failure immediately halts execution with an auditable terminal `FAILED` receipt.
+  - **Step 1 — Morning Brief**: Quantitative analysis, regime, conviction scores, gaps.
+  - **Step 2 — Triage**: Action queue generation, confluence checks, and evidence fingerprinting.
+  - **Step 3 — Action Cards**: Socratic, interactive presentation of individual trade/rebalance cards. Captures structured decision records (approval, trim, deferral, override reason, human actor). Interactive resume across process crashes is an explicit non-goal; crashed runs are marked `ABORTED`/`FAILED` and require a fresh run.
+  - **Step 4 — Self-Evolution**: Friction classification, learnings, and map-debt logging.
+  - **Step 5 — Summary & Cryptographic Finalization**: Executive rollup, report projection, and explicit terminal closure receipt.
+
+---
+
+## 2. Database Layer & Concurrency Architecture
+- **Schema Compatibility (Zero Column Migrations)**:
+  - Preserves existing `verification_receipts` columns (`receipt_id`, `task_id`, `gate_name`, `command_executed`, `exit_code`, `receipt_token`, `timestamp`).
+  - Structured envelope stored as canonical JSON in `receipt_token`:
+    ```json
+    {
+      "run_id": "DAILY-...",
+      "correlation_id": "...",
+      "step_num": 0,
+      "step_name": "READINESS",
+      "status": "COMPLETED",
+      "prev_chain_hash": "...",
+      "chain_hash": "...",
+      "payload": { ... }
+    }
+    ```
+- **Deterministic Gate Naming**:
+  - Step receipts: `gate_name = 'DAILY_RUN_' || run_id || '_STEP_' || step_num`.
+  - Terminal receipt: `gate_name = 'DAILY_RUN_' || run_id || '_TERMINAL'`.
+- **Enforcing Uniqueness & Write-Time Ordering**:
+  - A `UNIQUE INDEX IF NOT EXISTS idx_verification_receipts_gate_name ON verification_receipts(gate_name)` is created in `control_plane.db`. Duplicate inserts for any step or multiple terminal receipts fail instantly at the database engine level (`sqlite3.IntegrityError`).
+  - **Transaction Isolation**: Every receipt insert executes under `BEGIN IMMEDIATE` to acquire a write lock before querying `SELECT MAX(...)` for monotonic step verification (`step_num == last_step + 1`), preventing TOCTOU races between step writes and signal handlers.
+
+---
+
+## 3. Cryptographic Consistency & Realistic Threat Model
+- **Threat Model & Clarified Scope**:
+  - The hash chain provides **internal consistency, fault detection, and tamper detection against accidental modification or unauthorized script writes**. It does not claim resistance against a malicious superuser with raw SQL write access to `control_plane.db` without an external cryptographic anchor.
+  - As an external anchor, the terminal receipt's `final_chain_hash` is committed directly into the header of the generated daily markdown review (`data/history/reviews/daily/YYYY-MM-DD.md`), binding git history to the control plane.
+- **Canonical JSON & Float Normalization**:
+  - Forbids raw platform floats in receipt payloads. All floating-point quantities (prices, market values, conviction scores, weights) are normalized to fixed-precision strings (e.g. `f"{val:.4f}"` or `Decimal`) before hashing.
+  - `canonical_json()` enforces: sorted keys, fixed separators `(',', ':')`, UTF-8 encoding, and UTC ISO-8601 timestamps.
+- **Hash Chaining & Terminal Anchor**:
+  - `receipt_payload_hash = SHA256(canonical_json(normalized_payload))`
+  - `chain_hash_n = SHA256(chain_hash_{n-1} + ":" + receipt_payload_hash_n)` where `chain_hash_0 = SHA256(run_id)`.
+  - **Terminal Receipt (`DAILY_RUN_<run_id>_TERMINAL`)**:
+    - Commits: `run_id`, `mode`, `required_steps`, `final_step_hash`, `final_chain_hash`, `terminal_state`.
+    - Verifier strictly requires `DAILY_RUN_<run_id>_TERMINAL` to match the mode contract, rejecting any truncated prefix runs.
+
+---
+
+## 4. Sandboxed Directory Cleanup & Stale Run Janitor
+- **4-Way Bound Finalized Cleanup**:
+  - Scratch folder: `temp/daily_run_<run_id>/`.
+  - `clean_run_directory(run_dir, run_id, db_path)` enforces:
+    1. Unresolved path `Path(run_dir).is_symlink()` is `False`.
+    2. Path resolves to a direct child of `repo_root / "temp"` named `daily_run_<run_id>`.
+    3. Manifest `run_manifest.json` inside contains matching `run_id`.
+    4. Database record `DAILY_RUN_<run_id>_TERMINAL` exists with state `COMPLETED`.
+    5. Directory deleted using an atomic directory handle or leaf verification.
+- **Stale Run Janitor (`--prune-stale`)**:
+  - Reuses path guards 1, 2, and 3 (unresolved symlink check, direct child of `temp/`, name prefix check).
+  - **Process Liveness Check**: Verifies pid in `run_manifest.json` is no longer active (`os.kill(pid, 0)` raises `ProcessLookupError`) before pruning. Never prunes active or open sessions based on age alone.
+  - Records an auditable terminal `ABANDONED` receipt in `control_plane.db` prior to purging files.
+
+---
+
+## 5. Hardened Definition of Done (DoD)
+- [ ] `/daily` is the sole registered daily skill in `plugin.json` and instructions.
+- [ ] `/daily-brief` directory deleted with zero remaining references across skills, agents, and templates.
+- [ ] Unique index on `gate_name` created; duplicate step and duplicate terminal writes fail at insert time.
+- [ ] All database writes run under `BEGIN IMMEDIATE` with monotonic step validation (`step == last + 1`).
+- [ ] Numeric payloads normalized to fixed-precision strings; cross-environment canonical JSON hashing test passes.
+- [ ] Terminal receipt named `DAILY_RUN_<run_id>_TERMINAL` anchors run mode and final chain hash.
+- [ ] Unresolved path symlink check enforced prior to path resolution.
+- [ ] Janitor (`--prune-stale`) enforces path sandbox + process liveness check before pruning abandoned runs.
+- [ ] Step 0 verifies `control_plane.db` write transaction without coupling to git dev tooling.
+- [ ] Comprehensive adversarial unit tests pass in `plugins/portfolio-advisor/tests/test_daily_loop_receipts.py`.
+- [ ] Full regression suite (`python3 run_tests.py --unit`) passes cleanly.

plugins/portfolio-advisor/agents/daily-loop-agent.md

@@ -5,7 +5,6 @@ description: >
   portfolio freshness check → morning brief → interactive triage → action execution →
   self-evolution logging. One command replaces 10 manual steps. Compounds over time.
 dependencies:
-  - skill:daily-brief
   - skill:x-news-sweep
   - skill:rebalance-portfolio
   - skill:strategic-review

plugins/portfolio-advisor/plugin.json

@@ -74,11 +74,6 @@
             "path": "skills/13f-analyze/SKILL.md",
             "trigger": "/13f-analyze"
         },
-        {
-            "name": "daily_brief",
-            "path": "skills/daily-brief/SKILL.md",
-            "trigger": "/daily-brief"
-        },
         {
             "name": "daily_loop",
             "path": "skills/daily-loop/SKILL.md",
@@ -171,4 +166,4 @@
         "drift-detection",
         "currency-conversion"
     ]
-}
+}

plugins/portfolio-advisor/references/evolution-log.md

@@ -450,3 +450,10 @@ positions (vs the false 106.6h/0-positions reading before the fix).
 **Files patched:** `plugins/portfolio-advisor/agents/daily-loop-agent.md`
 (Step 0 script + readiness card + hard-gate text, now source-agnostic between
 TradingView and Questrade MCP sync paths).
+
+## 2026-09-06 — Daily Loop Consolidation & Hardened Control Plane Receipts
+- Consolidated morning operations into `/daily` (`--scan` vs `--interactive`).
+- Cleanly deleted `/daily-brief` to eliminate dual-command ambiguity.
+- Created `daily_receipts.py` with canonical JSON serialization, float normalization, and `BEGIN IMMEDIATE` transaction isolation with a UNIQUE index on `gate_name`.
+- Added explicit terminal closure receipt (`DAILY_RUN_<run_id>_TERMINAL`) to prevent prefix truncation attacks.
+- Upgraded `verify_daily_run.py` with 4-way cleanup sandboxing and process-liveness checking stale janitor.

plugins/portfolio-advisor/scripts/daily_receipts.py

@@ -0,0 +1,228 @@
+"""Deterministic receipt and hash-chaining engine for daily operations.
+
+Purpose:
+    Provides canonical JSON serialization with float normalization, SHA-256
+    hash-chaining, monotonic step validation under BEGIN IMMEDIATE SQLite
+    transactions, and database-level uniqueness enforcement for daily execution receipts.
+
+Layer:
+    plugins/portfolio-advisor/scripts (Execution & Audit Engine)
+
+Key Functions:
+    - ensure_receipt_index(con): Creates UNIQUE index on verification_receipts(gate_name).
+    - normalize_numerics(payload): Recursively converts floats to fixed-precision strings.
+    - canonical_json(data): Deterministic JSON serialization.
+    - compute_receipt_hash(payload): SHA-256 hash of normalized canonical JSON.
+    - compute_chain_hash(prev_hash, receipt_hash): Cryptographic hash accumulator.
+    - record_daily_receipt(db_path, run_id, step_num, step_name, status, payload):
+        Atomic monotonic step insertion.
+    - record_terminal_receipt(db_path, run_id, mode, state, required_steps, final_chain_hash):
+        Atomic terminal anchor insertion.
+"""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import os
+import secrets
+import sqlite3
+from datetime import datetime, timezone
+from decimal import Decimal
+from pathlib import Path
+from typing import Any, Dict, List, Optional
+
+
+class MonotonicOrderError(Exception):
+    """Raised when step insertion violates strict monotonic ordering."""
+
+
+class ReceiptError(Exception):
+    """Raised for general receipt engine failures."""
+
+
+def generate_run_id() -> str:
+    """Generate a control-plane owned unique run identifier."""
+    now_str = datetime.now(timezone.utc).strftime("%Y%m%d-%H%M%S")
+    rand_hex = secrets.token_hex(4).upper()
+    return f"DAILY-{now_str}-{rand_hex}"
+
+
+def ensure_receipt_index(con: sqlite3.Connection) -> None:
+    """Ensure unique index on verification_receipts(gate_name) exists."""
+    con.execute(
+        "CREATE UNIQUE INDEX IF NOT EXISTS idx_verification_receipts_gate_name "
+        "ON verification_receipts(gate_name)"
+    )
+
+
+def normalize_numerics(obj: Any) -> Any:
+    """Recursively convert float values to fixed-precision strings for cross-platform determinism."""
+    if isinstance(obj, float):
+        return f"{obj:.4f}"
+    elif isinstance(obj, Decimal):
+        return f"{obj:.4f}"
+    elif isinstance(obj, dict):
+        return {k: normalize_numerics(v) for k, v in obj.items()}
+    elif isinstance(obj, list):
+        return [normalize_numerics(v) for v in obj]
+    return obj
+
+
+def canonical_json(data: Any) -> str:
+    """Serialize data into deterministic, canonical JSON."""
+    return json.dumps(
+        data,
+        sort_keys=True,
+        separators=(",", ":"),
+        ensure_ascii=False,
+    )
+
+
+def compute_receipt_hash(payload: Dict[str, Any]) -> str:
+    """Compute SHA-256 hash of canonical JSON normalized payload."""
+    normalized = normalize_numerics(payload)
+    canonical = canonical_json(normalized)
+    return hashlib.sha256(canonical.encode("utf-8")).hexdigest()
+
+
+def compute_chain_hash(prev_chain_hash: str, receipt_hash: str) -> str:
+    """Compute chained accumulator hash: SHA256(prev_chain_hash + ':' + receipt_hash)."""
+    combined = f"{prev_chain_hash}:{receipt_hash}"
+    return hashlib.sha256(combined.encode("utf-8")).hexdigest()
+
+
+def record_daily_receipt(
+    db_path: Path | str,
+    run_id: str,
+    step_num: int,
+    step_name: str,
+    status: str,
+    payload: Dict[str, Any],
+    correlation_id: Optional[str] = None,
+) -> str:
+    """Record a daily step receipt under BEGIN IMMEDIATE transaction isolation."""
+    con = sqlite3.connect(str(db_path), isolation_level=None)
+    try:
+        ensure_receipt_index(con)
+        con.execute("BEGIN IMMEDIATE")
+
+        # Query existing steps for this run to enforce monotonic order & retrieve previous hash
+        cur = con.cursor()
+        cur.execute(
+            "SELECT receipt_token FROM verification_receipts "
+            "WHERE gate_name LIKE ? ORDER BY receipt_id ASC",
+            (f"DAILY_RUN_{run_id}_STEP_%",),
+        )
+        existing_rows = cur.fetchall()
+
+        if not existing_rows:
+            if step_num != 0:
+                raise MonotonicOrderError(
+                    f"First step must be 0 (Readiness), got step {step_num}"
+                )
+            prev_chain_hash = hashlib.sha256(run_id.encode("utf-8")).hexdigest()
+        else:
+            last_token_str = existing_rows[-1][0]
+            try:
+                last_envelope = json.loads(last_token_str)
+                last_step = int(last_envelope["step_num"])
+                prev_chain_hash = last_envelope["chain_hash"]
+            except Exception as e:
+                raise ReceiptError(f"Corrupt previous receipt token: {e}") from e
+
+            expected_step = last_step + 1
+            if step_num != expected_step:
+                raise MonotonicOrderError(
+                    f"Expected step {expected_step}, got step {step_num} (monotonic violation)"
+                )
+
+        receipt_hash = compute_receipt_hash(payload)
+        chain_hash = compute_chain_hash(prev_chain_hash, receipt_hash)
+
+        envelope = {
+            "run_id": run_id,
+            "correlation_id": correlation_id,
+            "step_num": step_num,
+            "step_name": step_name,
+            "status": status,
+            "prev_chain_hash": prev_chain_hash,
+            "receipt_hash": receipt_hash,
+            "chain_hash": chain_hash,
+            "payload": normalize_numerics(payload),
+            "timestamp": datetime.now(timezone.utc).isoformat(),
+        }
+
+        gate_name = f"DAILY_RUN_{run_id}_STEP_{step_num}"
+        receipt_token = canonical_json(envelope)
+
+        cur.execute(
+            "INSERT INTO verification_receipts "
+            "(task_id, gate_name, command_executed, exit_code, receipt_token) "
+            "VALUES (?, ?, ?, ?, ?)",
+            (
+                run_id,
+                gate_name,
+                f"step_{step_num}_{step_name.lower()}",
+                0 if status == "COMPLETED" else 1,
+                receipt_token,
+            ),
+        )
+        con.execute("COMMIT")
+        return chain_hash
+    except Exception:
+        con.execute("ROLLBACK")
+        raise
+    finally:
+        con.close()
+
+
+def record_terminal_receipt(
+    db_path: Path | str,
+    run_id: str,
+    mode: str,
+    state: str,
+    required_steps: List[int],
+    final_chain_hash: str,
+    error_msg: Optional[str] = None,
+    correlation_id: Optional[str] = None,
+) -> str:
+    """Record terminal anchor receipt for a daily run under BEGIN IMMEDIATE."""
+    con = sqlite3.connect(str(db_path), isolation_level=None)
+    try:
+        ensure_receipt_index(con)
+        con.execute("BEGIN IMMEDIATE")
+
+        gate_name = f"DAILY_RUN_{run_id}_TERMINAL"
+        envelope = {
+            "run_id": run_id,
+            "correlation_id": correlation_id,
+            "mode": mode,
+            "terminal_state": state,
+            "required_steps": required_steps,
+            "final_chain_hash": final_chain_hash,
+            "error_msg": error_msg,
+            "timestamp": datetime.now(timezone.utc).isoformat(),
+        }
+
+        receipt_token = canonical_json(envelope)
+        cur = con.cursor()
+        cur.execute(
+            "INSERT INTO verification_receipts "
+            "(task_id, gate_name, command_executed, exit_code, receipt_token) "
+            "VALUES (?, ?, ?, ?, ?)",
+            (
+                run_id,
+                gate_name,
+                "daily_run_finalize",
+                0 if state == "COMPLETED" else 1,
+                receipt_token,
+            ),
+        )
+        con.execute("COMMIT")
+        return receipt_token
+    except Exception:
+        con.execute("ROLLBACK")
+        raise
+    finally:
+        con.close()

plugins/portfolio-advisor/scripts/generate_reports.py

@@ -315,3 +315,27 @@ def main():
 
 if __name__ == "__main__":
     main()
+
+
+def render_interactive_decisions(decisions: list, final_chain_hash: str = "") -> str:
+    """Render interactive triage decisions into Markdown projection."""
+    lines = ["## Interactive Session Decisions & Overrides", ""]
+    if final_chain_hash:
+        lines.append(f"* **Receipt Anchor:** ")
+        lines.append("")
+    if not decisions:
+        lines.append("*(No interactive trade decisions or overrides recorded this session)*")
+        lines.append("")
+        return chr(10).join(lines)
+    lines.append("| Ticker | Recommended | Action Taken | Override Reason | Actor | Timestamp |")
+    lines.append("|---|---|---|---|---|---|")
+    for d in decisions:
+        t = d.get('ticker', 'N/A')
+        r = d.get('recommended', 'N/A')
+        a = d.get('action', 'N/A')
+        o = d.get('override_reason') or 'None'
+        act = d.get('actor', 'human')
+        ts = d.get('timestamp', 'N/A')
+        lines.append(f"| {t} | {r} | {a} | {o} | {act} | {ts} |")
+    lines.append("")
+    return chr(10).join(lines)