feat(FM.2): warm observe-only ambient monitor daemon

changelog.d/237.added.md

@@ -0,0 +1 @@
+- Ambient monitor daemon (FM.2): `doberman monitor run`/`status` score non-inline activity through the same engine, alert-only, never enforcing (#237, thanks @Maqbool61)

docs/CLI.md

@@ -27,6 +27,8 @@ Day-to-day posture, status, and review commands.
 | `doberman tui` | Interactive decision log with a plain-language "why" panel. Keys: `?` help (lists every binding, including the arrow keys/page up/page down that come from the table itself; a proper modal, since it never stacks, other keys are inert while it's open, and it also opens from inside the full-screen why view), `/` filter, `b`/`B`/`a` next/prev BLOCK or next AUTH (says "no other ... rows in view" rather than silently re-selecting the row you're already on when it's the only match), `w`/`enter` full-screen why, `tab` switch focus, `y` copy id, `home`/`end`, `r` reload, `q` quit. The footer always shows the same 6 bindings (`w`/`/`/`b`/`?`/`q`/`r`) at any width; `B`/`a`/`y` are keyboard-only, documented in `?`. `risk`/`auth` drop out of the table below 100 terminal columns (both are already restated in the why panel/full-screen why); the header reads `showing N of M (filtered) - X BLOCK / Y AUTH / Z PASS loaded` so it's clear which count is which. Times show in your local zone (`HH:MM:SS`, or date-qualified `MM-DD HH:MM` across a multi-day window); the why panel's border carries the row's absolute UTC instant plus a relative age ("2m ago", with a `[focus]` cue appended while it has focus), and, when its content overflows, a muted "(scroll for more)" cue in its border subtitle. The panel body itself always starts with the explanation. An empty log, a filtered-to-zero view, or a missing decision log replaces the table (not the why panel) with the message, in the table's own area. Below 76x16 the app shows a one-line "resize" notice instead of the browser, and its own footer then shows only `q`/"quit" (every key still works, just hidden). Needs the `tui` extra. | `--path`/`-p`, `--last`/`-n` |
 | `doberman dash` | Localhost-only dashboard: live decision feed, stats, and an AUTH approve/deny queue. Needs the `dash` extra. | `--port`, `--path`/`-p` |
 | `doberman demo` | Scripted attack reel through the real decision engine. Nothing runs against a real tool or downstream server. | `--path`/`-p`, `--mode`, `--fast`, `--quiet`/`-q` |
+| `doberman monitor run` | Warm, observe-only ambient daemon (FM.2): scores non-inline activity through the same engine, never enforces. Refuses to start a second instance for the same repo. | `--path`/`-p`, `--interval`, `--mode` |
+| `doberman monitor status` | Whether the ambient monitor daemon appears to be running, plus its bus cursor and pending-event backlog. | `--path`/`-p` |
 | `doberman revoke ELEVATION_ID` | Revoke an active role elevation by id (see `doberman status`). | `--path`/`-p` |
 | `doberman tune` | Friction report (interventions per session, top AUTH reasons) plus gated standing-elevation proposals. | `--path`/`-p`, `--json`, `--last`, `--min-occurrences`, `--accept` |
 | `doberman memory` | Learned-memory profile: decision counts, verdict mix, most-touched path classes. Never shows a fingerprint value or raw secret. | `--path`/`-p`, `--json` |

docs/REASON_CODES.md

@@ -1,6 +1,6 @@
 # Reason codes
 
-Every non-`PASS` Doberman decision carries one or more `ReasonCode` values plus a human explanation. `ReasonCode` itself is defined in `src/doberman/models.py`. This page catalogues every member of that enum: where in the code it actually gets attached to a decision, and what it means in plain language. The enum currently defines 64 codes. `doberman log` and `doberman tune --json` (see [CLI reference](CLI.md)) both surface these values directly, so matching on the name here is the stable way to script against a decision.
+Every non-`PASS` Doberman decision carries one or more `ReasonCode` values plus a human explanation. `ReasonCode` itself is defined in `src/doberman/models.py`. This page catalogues every member of that enum: where in the code it actually gets attached to a decision, and what it means in plain language. The enum currently defines 65 codes. `doberman log` and `doberman tune --json` (see [CLI reference](CLI.md)) both surface these values directly, so matching on the name here is the stable way to script against a decision.
 
 | Code | Group | Raised in | Meaning |
 |------|-------|-----------|---------|
@@ -9,6 +9,7 @@ Every non-`PASS` Doberman decision carries one or more `ReasonCode` values plus
 | `downstream_error` | General | `proxy/executor.py` | The downstream tool call, made over MCP (Model Context Protocol, the interface Doberman's proxy uses to reach real tools), raised while being forwarded, so the result is denied and the failure is never exposed to the agent. |
 | `single_use_elevation_unclaimable` | General | `proxy/executor.py` | A single-use elevation covered the action, but the atomic claim taken just before the forward found it already spent (a concurrent call won it) or could not be recorded, so the forward is refused rather than let two calls release on one approval. |
 | `effect_set_diverged` | General | `proxy/executor.py` | The proxy recomputes a delete-class command's bounded blast-radius effect set right before forwarding it. If that recompute's digest disagrees with what was shown to the human at approval time, in either direction, or a previously-known count becomes unknown, the forward is refused with a synthetic `BLOCK` rather than executed against a stale preview. |
+| `ambient_scoring_error` | General | `monitor/daemon.py` | The ambient monitor (FM.2) could not reconstruct a `SecurityObject` from a redacted `ActivityEvent`, or scoring it through `decide()` raised. Per-event isolation records a conservative alert row instead of dropping the event silently or letting one poisoned event kill the daemon's tick loop. Never produced on the live/inline decision path. |
 | `objective_guardrail_error` | General | `engine/decision_engine.py`, `proxy/executor.py` | The objective guardrail, or a post-decision integrity check such as the output secret scan or artifact digest check, raised an unhandled exception, so the action fails closed to `BLOCK`. |
 | `subjective_guardrail_error` | General | `engine/decision_engine.py`, `engine/subjective.py` | The subjective guardrail raised an unhandled exception after the objective guardrail already passed, so the action escalates to authentication instead of silently passing. |
 | `subjective_block_clamped` | General | `engine/decision_engine.py`, `policy/drift.py` | The subjective guardrail returned `BLOCK`, but none of its reason codes sit on the hard-block allowlist, so the execution rule clamps that `BLOCK` down to `AUTH` and records that the clamp happened. |

src/doberman/cli/main.py

@@ -90,6 +90,7 @@
 from doberman.render import (
     format_utc_timestamp,
     humanize_auth_result,
+    is_ambient_source_context,
     next_step_line,
     style_text,
     verdict_label,
@@ -187,6 +188,15 @@ def _ensure_encode_safe_stdio() -> None:
 )
 app.add_typer(plugins_app, name="plugins", rich_help_panel="Advanced")
 
+monitor_app = typer.Typer(
+    help=(
+        "Ambient observe-only daemon (FM.2): scores non-inline activity through "
+        "the same decision engine, never enforces."
+    ),
+    no_args_is_help=True,
+)
+app.add_typer(monitor_app, name="monitor", rich_help_panel="Advanced")
+
 memory_app = typer.Typer(
     help="Learned behavioral memory: profile, gated reset, and retention pruning.",
     no_args_is_help=False,
@@ -1228,7 +1238,7 @@ def _status_payload(path: str) -> dict:
     """Collect the same redacted data the text and JSON status views share.
 
     Nothing secret-shaped is included: enrollment is a boolean, elevations carry
-    ids/scopes/expiry only, decisions carry ts/verdict/reason codes only.
+    ids/scopes/expiry only, decisions carry ts/verdict/reason codes/source_context only.
     """
     from doberman.storage.policy_catalogue import current_snapshot, policy_version
 
@@ -1277,6 +1287,7 @@ def _status_payload(path: str) -> dict:
                 "ts": row["ts"],
                 "final_verdict": row["final_verdict"],
                 "reason_codes": reasons,
+                "source_context": row["source_context"],
             }
         )
 
@@ -1431,7 +1442,10 @@ def _render_status_text(payload: dict) -> None:
     else:
         for row in recent:
             reasons = ", ".join(row["reason_codes"]) or "-"
-            typer.echo(f"  {row['ts']}  {verdict_label_str(row['final_verdict'])}  {reasons}")
+            ambient = is_ambient_source_context(row.get("source_context"))
+            typer.echo(
+                f"  {row['ts']}  {verdict_label_str(row['final_verdict'], ambient=ambient)}  {reasons}"
+            )
 
     missed = payload["missed_challenges_24h"]
     if missed:
@@ -2327,7 +2341,14 @@ def tune(
 # for "allowed without a human" cannot answer the question #399 raised. Both are
 # closed values (an AuthPath enum member; 1/0/NULL) and can never carry command
 # text, so exporting them widens the stream by nothing an operator must redact.
-_JSONL_EXTRA_COLUMNS = ("id", "agent_role", "risk", "auth_path", "human_confirmed")
+_JSONL_EXTRA_COLUMNS = (
+    "id",
+    "agent_role",
+    "risk",
+    "auth_path",
+    "human_confirmed",
+    "source_context",
+)
 
 # Keep every action type in one column even when a new enum member outgrows the
 # historical 13-character values (network_request/package_install are 15).
@@ -2394,18 +2415,24 @@ def log(
     for row in rows:
         target = row["target_path_class"] or "-"
         reasons = ", ".join(json.loads(row["reason_codes_json"] or "[]")) or "-"
+        ambient = is_ambient_source_context(row.get("source_context"))
         # A pending AUTH row (no answer yet) must never look identical to "no
-        # auth step at all" - round 5 design critique item 7.
+        # auth step at all" - round 5 design critique item 7. An ambient
+        # AUTH-grade alert (FM.2) was never actually challenged, so it must
+        # never show "pending - not yet answered" either - that reads as a
+        # live outstanding challenge, exactly what the hard rule ("no output
+        # may read as blocked") forbids.
         auth = (
             f"; auth={humanize_auth_result(row['auth_result'], verdict=row['final_verdict'])}"
-            if row["auth_result"] or row["final_verdict"] == "AUTH"
+            if not ambient and (row["auth_result"] or row["final_verdict"] == "AUTH")
             else ""
         )
         # round 8 design critique item 7: the same "YYYY-MM-DD HH:MM:SS UTC"
         # format the tui's why panel shows (no microseconds) - `--jsonl`
         # keeps the raw stored `ts` string unchanged (scripts parse that one).
         typer.echo(
-            f"{format_utc_timestamp(row['ts'])}  {verdict_label_str(row['final_verdict'])} "
+            f"{format_utc_timestamp(row['ts'])}  "
+            f"{verdict_label_str(row['final_verdict'], ambient=ambient)} "
             f"{row['action_type']:<{_ACTION_WIDTH}} {target}  [{reasons}]{auth}"
         )
         # --why (round 4 design critique item 8, round 6 item 7): a compact,
@@ -2420,7 +2447,7 @@ def log(
             any_explained = True
             for line in wrap_detail(why_body(row)):
                 typer.echo(line)
-            next_line = next_step_line(row["final_verdict"], tui_hint=False)
+            next_line = next_step_line(row["final_verdict"], tui_hint=False, ambient=ambient)
             if next_line:
                 for line in wrap_detail(next_line):
                     typer.echo(line)
@@ -2572,6 +2599,92 @@ def demo(
         raise typer.Exit(code=1)
 
 
+@monitor_app.command("run")
+def monitor_run(
+    path: str = typer.Option(".", "--path", "-p", help="Repository root to observe."),
+    interval: float = typer.Option(
+        5.0, "--interval", min=0.5, help="Seconds between scoring ticks."
+    ),
+    mode: str = typer.Option(
+        "balanced", "--mode", help="Security mode used to score ambient events."
+    ),
+) -> None:
+    """Run the warm, observe-only ambient daemon (FM.2, issue #237).
+
+    Each tick, polls every registered `doberman.collectors` entry point,
+    emits what they see onto the FM.1 activity bus, drains the bus from this
+    daemon's own saved cursor, and scores each drained event through the
+    SAME decision engine the live gate uses -- recording an alert row for
+    anything AUTH/BLOCK-grade. Structurally observe-only: no prompter, no
+    executor, and no challenge is ever imported or constructed here, so
+    nothing this process does can block, challenge, or execute anything. A
+    dead or never-started monitor changes nothing about the live inline gate
+    -- it is a second, independent consumer of the bus, not a dependency of
+    the decision path.
+
+    Refuses to start a second daemon for the same `--path` while another
+    one's heartbeat is still fresh (see `doberman monitor status`). Runs
+    until interrupted with Ctrl+C.
+    """
+    from doberman.monitor.daemon import (
+        MONITOR_HEARTBEAT_MAX_AGE_S,
+        MonitorAlreadyRunning,
+        run_forever,
+    )
+
+    try:
+        resolved_mode = resolve_mode(mode)
+    except ValueError as exc:
+        typer.echo(f"error: {exc}", err=True)
+        raise typer.Exit(code=1) from exc
+
+    typer.echo(
+        f"doberman monitor: observe-only, ticking every {interval:.1f}s "
+        f"(mode={resolved_mode.value})"
+    )
+    typer.echo("Ctrl+C to stop. This process never blocks, challenges, or executes anything.")
+
+    try:
+        run_forever(path, interval_s=interval, mode=resolved_mode.value)
+    except MonitorAlreadyRunning:
+        typer.echo(
+            "error: a doberman monitor daemon already appears to be running for "
+            f"this repo (heartbeat fresher than {MONITOR_HEARTBEAT_MAX_AGE_S:.0f}s); "
+            "not starting a second one",
+            err=True,
+        )
+        raise typer.Exit(code=1) from None
+    except KeyboardInterrupt:
+        typer.echo("\ndoberman monitor: stopped.")
+
+
+@monitor_app.command("status")
+def monitor_status_cmd(
+    path: str = typer.Option(".", "--path", "-p", help="Repository root to report on."),
+) -> None:
+    """Report whether the ambient monitor daemon (FM.2) appears to be running.
+
+    Shows the last heartbeat age, the daemon's saved bus cursor, and how many
+    activity-bus events are still waiting to be drained -- a growing pending
+    count with no running daemon usually means the daemon isn't started; a
+    growing count WITH one running usually means a collector is producing
+    events faster than the daemon can score them.
+    """
+    import asyncio
+
+    from doberman.monitor.daemon import monitor_status
+
+    status = asyncio.run(monitor_status(path))
+    state = "RUNNING" if status["running"] else "NOT RUNNING"
+    typer.echo(f"doberman monitor: {state}")
+    if status["heartbeat_age_s"] is not None:
+        typer.echo(f"  last heartbeat: {status['heartbeat_age_s']:.1f}s ago")
+    else:
+        typer.echo("  last heartbeat: never")
+    typer.echo(f"  cursor: {status['cursor']}")
+    typer.echo(f"  pending events: {status['pending_events']}")
+
+
 @memory_app.callback(invoke_without_command=True, rich_help_panel="Policy internals")
 def memory(
     ctx: typer.Context,

src/doberman/explain.py

@@ -144,6 +144,10 @@
         "the recomputed blast radius for this delete no longer matches what was shown "
         "at approval time"
     ),
+    "ambient_scoring_error": (
+        "the ambient monitor could not score this observation cleanly, so it recorded "
+        "a conservative alert row instead of dropping it silently"
+    ),
     "correlated_trifecta": (
         "this action, combined with earlier ones in the session, adds up to the "
         "lethal-trifecta pattern"
@@ -248,6 +252,26 @@ def _describe_reason(code: str) -> str:
     return REASON_DESCRIPTIONS.get(code, code.replace("_", " "))
 
 
+#: FM.2: the ambient monitor (doberman.monitor.daemon) tags every row it
+#: writes via `storage.log.record_decision`'s `source_context_override` as
+#: `"ambient:<collector_id>"` - a shape no other writer produces (the live
+#: gate always writes a real `SourceContext` enum value: "user",
+#: "tool_output", "unknown", ...). Detecting on this column, rather than a
+#: new one, keeps the ambient/inline distinction visible everywhere a row's
+#: `source_context` already flows (the dash feed, the LLM narrator's
+#: allowlisted payload, `doberman log --jsonl`) with no schema change.
+_AMBIENT_SOURCE_PREFIX = "ambient:"
+#: FM.2 hard rule: every rendered explanation for an ambient row leads with
+#: this, so it can never be mistaken for a live enforcement outcome.
+_AMBIENT_EXPLANATION_PREFIX = "observed (not enforced): "
+
+
+def _is_ambient_row(row: dict) -> bool:
+    """Whether `row` was produced by the ambient monitor (FM.2), not the live gate."""
+    source_context = row.get("source_context")
+    return isinstance(source_context, str) and source_context.startswith(_AMBIENT_SOURCE_PREFIX)
+
+
 def _layer_checked_clause(layer: str) -> str:
     """What Doberman checked, in plain words (round 4 design critique item 6):
     "checking the rules" for the objective layer alone, or "...and the
@@ -267,7 +291,8 @@ def first_sentence(row: dict) -> str:
     """
     verdict = row.get("final_verdict") or "UNKNOWN"
     layer = row.get("decided_layer") or "objective"
-    return f"Doberman decided {verdict} after {_layer_checked_clause(layer)}."
+    sentence = f"Doberman decided {verdict} after {_layer_checked_clause(layer)}."
+    return f"{_AMBIENT_EXPLANATION_PREFIX}{sentence}" if _is_ambient_row(row) else sentence
 
 
 def _body_sentences(row: dict, *, with_reasons: bool = True) -> list[str]:
@@ -313,7 +338,16 @@ def _body_sentences(row: dict, *, with_reasons: bool = True) -> list[str]:
         else:
             sentences.append("No specific reason codes were recorded for this decision.")
 
-    if verdict == "AUTH":
+    if _is_ambient_row(row) and verdict in ("AUTH", "BLOCK"):
+        # FM.2 hard rule: an ambient AUTH/BLOCK-grade verdict is an alert
+        # row, nothing more - no output may read as if this action was
+        # actually blocked or challenged. The inline gate (not this daemon)
+        # remains the only enforcement point, dead or alive.
+        sentences.append(
+            "This is an ambient observation only - nothing was blocked or challenged; "
+            "the inline gate remains the sole enforcement point."
+        )
+    elif verdict == "AUTH":
         sentences.append(
             "Completing the authentication challenge (or an approved role elevation) "
             "would let this action proceed."
@@ -340,7 +374,8 @@ def template_explanation(row: dict, *, with_reasons: bool = True) -> str:
         *_body_sentences(row, with_reasons=with_reasons),
         f"(Checked by: {_describe_checked_by(layer)}.)",
     ]
-    return " ".join(sentences)
+    text = " ".join(sentences)
+    return f"{_AMBIENT_EXPLANATION_PREFIX}{text}" if _is_ambient_row(row) else text
 
 
 def why_body(row: dict) -> str:
@@ -352,7 +387,8 @@ def why_body(row: dict) -> str:
     what was already on screen; this adds the "what was attempted" and
     "Reasons: ..." sentences too, so `--why` earns its name.
     """
-    return " ".join(_body_sentences(row))
+    text = " ".join(_body_sentences(row))
+    return f"{_AMBIENT_EXPLANATION_PREFIX}{text}" if _is_ambient_row(row) else text
 
 
 #: Verdict -> the word :func:`headline` uses for what happened. Deliberately
@@ -461,7 +497,12 @@ def headline(row: dict) -> str:
     """
     reason_codes = _parse_reason_codes(row.get("reason_codes_json"))
     verdict = row.get("final_verdict") or "UNKNOWN"
-    verdict_word = _HEADLINE_VERDICT_WORD.get(verdict, verdict.lower())
+    if _is_ambient_row(row) and verdict in ("AUTH", "BLOCK"):
+        # FM.2: an ambient AUTH/BLOCK-grade verdict is an alert, never a
+        # claim that the action was actually stopped or challenged.
+        verdict_word = "flagged (not enforced)"
+    else:
+        verdict_word = _HEADLINE_VERDICT_WORD.get(verdict, verdict.lower())
     action_type = row.get("action_type") or "action"
     target = row.get("target_path_class")