diff --git a/assets/sourceos/bin/turtle-netwatch b/assets/sourceos/bin/turtle-netwatch
new file mode 100755
index 00000000000..f6808466ddf
--- /dev/null
+++ b/assets/sourceos/bin/turtle-netwatch
@@ -0,0 +1,396 @@
+#!/usr/bin/env python3
+"""turtle-netwatch — the Network/Connections agent for SourceOS.
+
+The one built-in agent the Agent-First Node Architecture was missing: an
+OS-native observer of sockets / DNS / process ownership that
+
+  1. OBSERVES  current connections (portable: `ss` on Linux, `lsof` on macOS/BSD)
+  2. EMITS     agent.v1 `Observation` events (schemas/agent/observation.avsc)
+  3. GRAPHS    them into a System Graph subgraph (KnowledgeUpdate deltas) for
+               hellgraph's AtomSpace (nodes: Process/Host/Port/User; edges:
+               CONNECTS_TO / OWNED_BY)
+  4. DETECTS   policy-bound anomalies (beaconing, egress spikes)
+  5. PROPOSES  an `Action` (block-domain / throttle-process) that is
+               ADMITTED-OR-REFUSED by the consent plane (a network mutation is an
+               operate/egress purpose) and then routed to the Governor for
+               human approval — nothing is applied on a deny.
+
+Binds, does not rebuild: hosts under turtle-agentd, gates through the
+consent-plane engine (policy-fabric purpose_admissibility_gate), and routes
+approvals to guardrail-fabric. Its whole operation ships as a turtle-runbook
+(runbooks/netwatch.yaml) so a user OR an agent can execute it step by step.
+
+  turtle-netwatch snapshot [--json]
+  turtle-netwatch observe  [--window SEC] [--interval SEC] [--json]
+  turtle-netwatch graph    [--from FILE] [--json]
+  turtle-netwatch detect   [--from FILE] [--json]
+  turtle-netwatch propose  --action block-domain|throttle-process --target X [--apply] [--json]
+"""
+from __future__ import annotations
+
+import argparse
+import datetime as dt
+import hashlib
+import json
+import os
+import platform
+import re
+import shutil
+import statistics
+import subprocess
+import sys
+import time
+from pathlib import Path
+from typing import Any
+
+SCHEMA_NS = "agent.v1"
+
+
+def utc_now() -> str:
+    return dt.datetime.now(dt.timezone.utc).isoformat().replace("+00:00", "Z")
+
+
+def state_dir() -> Path:
+    base = os.environ.get(
+        "SOURCEOS_TERMINAL_RECEIPTS",
+        str(Path.home() / ".local" / "state" / "sourceos" / "terminal" / "receipts"),
+    )
+    d = Path(base).parent / "netwatch"
+    d.mkdir(parents=True, exist_ok=True)
+    return d
+
+
+def _seal(obj: Any) -> str:
+    return hashlib.sha256(json.dumps(obj, sort_keys=True, default=str).encode()).hexdigest()[:16]
+
+
+# --------------------------------------------------------------------------- observe
+_EXTERNAL_DENY = ("127.", "0.0.0.0", "::1", "*", "localhost")
+
+
+def _is_external(addr: str) -> bool:
+    a = addr.strip("[]")
+    if not a or a.startswith(_EXTERNAL_DENY):
+        return False
+    if a.startswith(("10.", "192.168.", "169.254.", "fe80:")):
+        return False
+    if re.match(r"^172\.(1[6-9]|2\d|3[01])\.", a):
+        return False
+    return True
+
+
+def snapshot() -> list[dict[str, Any]]:
+    """Portable connection snapshot. Returns normalized connection records."""
+    if shutil.which("ss"):
+        return _parse_ss()
+    if shutil.which("lsof"):
+        return _parse_lsof()
+    return []
+
+
+def _parse_ss() -> list[dict[str, Any]]:
+    try:
+        out = subprocess.run(
+            ["ss", "-tunp"], text=True, capture_output=True, timeout=10
+        ).stdout
+    except Exception:
+        return []
+    conns = []
+    for line in out.splitlines()[1:]:
+        f = line.split()
+        if len(f) < 5:
+            continue
+        proto, state, local, peer = f[0], f[1], f[-3], f[-2]
+        proc = ""
+        m = re.search(r'users:\(\("([^"]+)",pid=(\d+)', line)
+        pid, name = (m.group(2), m.group(1)) if m else ("", "")
+        raddr, _, rport = peer.rpartition(":")
+        conns.append(_conn(proto, local, raddr, rport, state, pid, name))
+    return conns
+
+
+def _parse_lsof() -> list[dict[str, Any]]:
+    try:
+        out = subprocess.run(
+            ["lsof", "-nP", "-i"], text=True, capture_output=True, timeout=10
+        ).stdout
+    except Exception:
+        return []
+    conns = []
+    for line in out.splitlines()[1:]:
+        f = line.split()
+        if len(f) < 9:
+            continue
+        name, pid, proto, node = f[0], f[1], f[7], f[8]
+        if "->" not in node:
+            continue
+        local, _, peer = node.partition("->")
+        raddr, _, rport = peer.rpartition(":")
+        state = f[9].strip("()") if len(f) > 9 else ""
+        conns.append(_conn(proto, local, raddr, rport, state, pid, name))
+    return conns
+
+
+def _conn(proto, local, raddr, rport, state, pid, name) -> dict[str, Any]:
+    return {
+        "ts": utc_now(),
+        "proto": proto.lower(),
+        "laddr": local,
+        "raddr": raddr,
+        "rport": rport,
+        "state": state,
+        "pid": pid,
+        "process": name,
+        "external": _is_external(raddr),
+    }
+
+
+def observation(conn: dict[str, Any]) -> dict[str, Any]:
+    """agent.v1 Observation for one connection."""
+    sev = "WARN" if conn["external"] and conn["proto"] == "tcp" else "INFO"
+    return {
+        "schema": f"{SCHEMA_NS}.Observation",
+        "source": "netwatch",
+        "type": "net.conn",
+        "ts": conn["ts"],
+        "attrs": {
+            "proto": conn["proto"],
+            "raddr": conn["raddr"],
+            "rport": str(conn["rport"]),
+            "process": conn["process"],
+            "pid": str(conn["pid"]),
+            "state": conn["state"],
+            "external": str(conn["external"]).lower(),
+        },
+        "severity": sev,
+    }
+
+
+def cmd_observe(args) -> int:
+    end = time.time() + args.window
+    seen, obs = set(), []
+    while True:
+        for c in snapshot():
+            key = (c["proto"], c["raddr"], c["rport"], c["pid"])
+            if key not in seen:
+                seen.add(key)
+                obs.append(observation(c))
+        if time.time() >= end:
+            break
+        time.sleep(args.interval)
+    out = state_dir() / "observations.jsonl"
+    with out.open("a") as fh:
+        for o in obs:
+            fh.write(json.dumps(o) + "\n")
+    result = {"emitted": len(obs), "sink": str(out), "window_s": args.window}
+    _print(args, result if not args.json else obs)
+    return 0
+
+
+# --------------------------------------------------------------------------- graph
+def build_system_graph(obs: list[dict[str, Any]]) -> dict[str, Any]:
+    """Project observations to a System Graph subgraph as a KnowledgeUpdate
+    delta (nodes + edges), ready for hellgraph AtomSpace ingestion."""
+    nodes: dict[str, dict] = {}
+    edges: list[dict] = []
+
+    def node(nid, kind, **attrs):
+        nodes.setdefault(nid, {"id": nid, "kind": kind, "attrs": attrs})
+
+    for o in obs:
+        a = o["attrs"]
+        proc = f"process:{a.get('process','?')}#{a.get('pid','?')}"
+        host = f"host:{a.get('raddr','?')}"
+        port = f"port:{a.get('rport','?')}/{a.get('proto','?')}"
+        node(proc, "Process", pid=a.get("pid"), name=a.get("process"))
+        node(host, "Host", external=a.get("external"))
+        node(port, "Port", proto=a.get("proto"))
+        edges.append({"from": proc, "rel": "CONNECTS_TO", "to": host, "via": port,
+                      "severity": o.get("severity", "INFO"), "ts": o["ts"]})
+    return {
+        "schema": f"{SCHEMA_NS}.KnowledgeUpdate",
+        "graph": "SYSTEM",
+        "ts": utc_now(),
+        "patch": {"nodes": list(nodes.values()), "edges": edges},
+        "prov": {"source": "netwatch", "seal": _seal({"n": sorted(nodes), "e": edges})},
+    }
+
+
+def _load_obs(args) -> list[dict[str, Any]]:
+    src = Path(args.__dict__.get("from") or (state_dir() / "observations.jsonl"))
+    if not src.exists():
+        return []
+    return [json.loads(l) for l in src.read_text().splitlines() if l.strip()]
+
+
+def cmd_graph(args) -> int:
+    ku = build_system_graph(_load_obs(args))
+    out = state_dir() / "system_graph.json"
+    out.write_text(json.dumps(ku, indent=2))
+    _print(args, ku if args.json else {"nodes": len(ku["patch"]["nodes"]),
+                                       "edges": len(ku["patch"]["edges"]), "sink": str(out)})
+    return 0
+
+
+# --------------------------------------------------------------------------- detect
+def detect_anomalies(obs: list[dict[str, Any]]) -> list[dict[str, Any]]:
+    """Beaconing (low-variance periodic contact to one dst) + egress fan-out."""
+    findings = []
+    by_dst: dict[tuple, list[str]] = {}
+    ext_hosts: set[str] = set()
+    for o in obs:
+        a = o["attrs"]
+        dst = (a.get("raddr"), a.get("rport"), a.get("process"))
+        by_dst.setdefault(dst, []).append(o["ts"])
+        if a.get("external") == "true":
+            ext_hosts.add(a.get("raddr"))
+    # beaconing: >=4 contacts to same dst with low coefficient of variation
+    for (raddr, rport, proc), times in by_dst.items():
+        if len(times) < 4:
+            continue
+        ep = [dt.datetime.fromisoformat(t.replace("Z", "+00:00")).timestamp() for t in sorted(times)]
+        gaps = [b - a for a, b in zip(ep, ep[1:])]
+        if len(gaps) >= 3 and statistics.mean(gaps) > 0:
+            cv = statistics.pstdev(gaps) / statistics.mean(gaps)
+            if cv < 0.20:
+                findings.append(_finding(
+                    "CRIT", "net.beaconing",
+                    f"{proc} beacons to {raddr}:{rport} every ~{statistics.mean(gaps):.0f}s (cv={cv:.2f})",
+                    {"raddr": raddr, "rport": rport, "process": proc, "count": str(len(times))}))
+    # egress fan-out: one process reaching many external hosts
+    proc_ext: dict[str, set] = {}
+    for o in obs:
+        a = o["attrs"]
+        if a.get("external") == "true":
+            proc_ext.setdefault(a.get("process"), set()).add(a.get("raddr"))
+    for proc, hosts in proc_ext.items():
+        if len(hosts) >= 10:
+            findings.append(_finding(
+                "WARN", "net.egress_fanout",
+                f"{proc} reached {len(hosts)} external hosts (possible exfil/scan)",
+                {"process": proc, "host_count": str(len(hosts))}))
+    return findings
+
+
+def _finding(sev, kind, msg, attrs) -> dict[str, Any]:
+    return {"schema": f"{SCHEMA_NS}.Observation", "source": "netwatch", "type": kind,
+            "ts": utc_now(), "severity": sev, "message": msg, "attrs": attrs}
+
+
+def cmd_detect(args) -> int:
+    findings = detect_anomalies(_load_obs(args))
+    _print(args, findings if args.json else
+           {"findings": len(findings), "detail": [f["message"] for f in findings]})
+    return 0 if not findings else 2
+
+
+# --------------------------------------------------------------------------- propose (consent-gated)
+def _consent_enforce(request: dict[str, Any]) -> tuple[bool, list[str], str]:
+    """Gate a network Action through the consent plane. Binds to the real
+    policy-fabric engine when a sibling checkout is present; FAILS CLOSED (deny)
+    otherwise — a network mutation must never proceed ungated."""
+    root = os.environ.get("PROPHET_POLICY_FABRIC")
+    candidates = [root] if root else []
+    for p in Path(__file__).resolve().parents:
+        candidates.append(str(p / "policy-fabric"))
+    for c in candidates:
+        pf = Path(c) if c else None
+        if pf and (pf / "policy_fabric" / "purpose_admissibility_gate.py").exists():
+            sys.path.insert(0, str(pf))
+            try:
+                from policy_fabric import purpose_admissibility_gate as gate  # type: ignore
+                doc = gate.decide(request, gate.load_catalogs())
+                d = doc["spec"]
+                return d["decision"] == "admit", d.get("denyReasons", []), doc["metadata"]["name"]
+            except Exception as exc:  # fail closed
+                return False, [f"consent engine error: {exc} (fail-closed)"], "gate-error"
+    return False, ["consent-plane engine not found (fail-closed): set $PROPHET_POLICY_FABRIC"], "no-gate"
+
+
+_ACTIONS = {
+    "block-domain": {"capability": "net.block", "tool": "exec-mutate", "purpose": "operate"},
+    "throttle-process": {"capability": "net.throttle", "tool": "exec-mutate", "purpose": "operate"},
+}
+
+
+def cmd_propose(args) -> int:
+    spec = _ACTIONS.get(args.action)
+    if not spec:
+        print(f"unknown action {args.action!r}; choose from {list(_ACTIONS)}", file=sys.stderr)
+        return 1
+    request = {
+        "role": "operator", "surface": "cluster-operator", "space": "system-space",
+        "tool": spec["tool"], "declaredPurpose": spec["purpose"],
+        "consent": {"purposes": [spec["purpose"]]},
+        "subjectRef": "urn:agent:netwatch",
+    }
+    admitted, reasons, gate_name = _consent_enforce(request)
+    action = {
+        "schema": f"{SCHEMA_NS}.Action", "capability": spec["capability"],
+        "args": {"target": args.target}, "ts": utc_now(),
+        "policy": {"space": "system-space", "purpose": spec["purpose"], "gate": gate_name},
+    }
+    if not admitted:
+        receipt = {"kind": "netwatch.action.refused", "action": action,
+                   "denyReasons": reasons, "ts": utc_now()}
+        _write_receipt(receipt)
+        _print(args, {"decision": "deny", "denyReasons": reasons, "receipt": receipt["kind"]})
+        print("REFUSED (fail-closed): network action not admissible.", file=sys.stderr)
+        return 3
+    # admitted -> route to Governor (guardrail-fabric) for human approval; do NOT
+    # auto-apply. Applying requires --apply AND an approval token from the Governor.
+    approval = os.environ.get("NETWATCH_GOVERNOR_APPROVAL")
+    receipt = {"kind": "netwatch.action.proposed", "action": action,
+               "governor": "guardrail-fabric", "approved": bool(approval and args.apply),
+               "seal": _seal(action), "ts": utc_now()}
+    _write_receipt(receipt)
+    status = "applied" if receipt["approved"] else "awaiting-governor-approval"
+    _print(args, {"decision": "admit", "status": status, "action": action["capability"],
+                  "target": args.target, "receipt": receipt["kind"]})
+    return 0
+
+
+def _write_receipt(receipt: dict[str, Any]) -> None:
+    out = state_dir() / "actions.jsonl"
+    with out.open("a") as fh:
+        fh.write(json.dumps(receipt) + "\n")
+
+
+# --------------------------------------------------------------------------- cli
+def cmd_snapshot(args) -> int:
+    conns = snapshot()
+    _print(args, conns if args.json else {"connections": len(conns),
+           "external": sum(1 for c in conns if c["external"]),
+           "collector": "ss" if shutil.which("ss") else ("lsof" if shutil.which("lsof") else "none"),
+           "os": platform.system()})
+    return 0
+
+
+def _print(args, obj) -> None:
+    print(json.dumps(obj, indent=2))
+
+
+def main(argv=None) -> int:
+    p = argparse.ArgumentParser(prog="turtle-netwatch", description="Network/Connections agent")
+    sub = p.add_subparsers(dest="cmd", required=True)
+    for name in ("snapshot", "graph", "detect"):
+        s = sub.add_parser(name)
+        s.add_argument("--json", action="store_true")
+        if name in ("graph", "detect"):
+            s.add_argument("--from", dest="from", default=None)
+    o = sub.add_parser("observe")
+    o.add_argument("--window", type=int, default=5)
+    o.add_argument("--interval", type=int, default=1)
+    o.add_argument("--json", action="store_true")
+    pr = sub.add_parser("propose")
+    pr.add_argument("--action", required=True, choices=list(_ACTIONS))
+    pr.add_argument("--target", required=True)
+    pr.add_argument("--apply", action="store_true")
+    pr.add_argument("--json", action="store_true")
+    args = p.parse_args(argv)
+    return {"snapshot": cmd_snapshot, "observe": cmd_observe, "graph": cmd_graph,
+            "detect": cmd_detect, "propose": cmd_propose}[args.cmd](args)
+
+
+if __name__ == "__main__":
+    sys.exit(main())
diff --git a/assets/sourceos/runbooks/netwatch.yaml b/assets/sourceos/runbooks/netwatch.yaml
new file mode 100644
index 00000000000..58dec4653d3
--- /dev/null
+++ b/assets/sourceos/runbooks/netwatch.yaml
@@ -0,0 +1,20 @@
+name: netwatch
+description: Network/Connections agent — observe → graph → detect → consent-gated propose
+# Executable by a user (`turtle-runbook run netwatch`) or an agent (each step is a
+# plain command with a deterministic, inspectable effect). Every mutating action
+# is refused fail-closed unless the consent plane admits it AND the Governor
+# (guardrail-fabric) approves — so this runbook is safe to run and to hand to an
+# agent.
+steps:
+  - cmd: turtle-netwatch snapshot
+    desc: "Point-in-time connection snapshot (portable: ss on Linux, lsof on macOS). Verifies a collector is present."
+  - cmd: turtle-netwatch observe --window 15 --interval 3
+    desc: "Observe for 15s; emit agent.v1 Observation events to the netwatch sink (schemas/agent/observation.avsc)."
+  - cmd: turtle-netwatch graph
+    desc: "Project the observations into a System Graph subgraph (KnowledgeUpdate delta) for hellgraph AtomSpace ingestion."
+  - cmd: turtle-netwatch detect
+    desc: "Scan the window for anomalies — beaconing (low-variance periodic contact) and egress fan-out. Exit 2 if any found."
+  - cmd: turtle-netwatch propose --action block-domain --target suspicious.example
+    desc: "Propose a network Action. It is ADMITTED-OR-REFUSED by the consent plane (operate purpose, system-space); on admit it is routed to the Governor for approval and NOT auto-applied. Deny => refused fail-closed."
+  - cmd: cat "${SOURCEOS_TERMINAL_RECEIPTS:-$HOME/.local/state/sourceos/terminal/receipts}/../netwatch/actions.jsonl"
+    desc: "Review the action receipts (proposed / refused) — every decision is hash-sealed and auditable."
diff --git a/assets/sourceos/schemas/agent/action.avsc b/assets/sourceos/schemas/agent/action.avsc
new file mode 100644
index 00000000000..c64d8703e4d
--- /dev/null
+++ b/assets/sourceos/schemas/agent/action.avsc
@@ -0,0 +1,11 @@
+{
+  "type": "record", "name": "Action", "namespace": "agent.v1",
+  "doc": "A proposed effect (e.g. net.block, net.throttle). Admissible ONLY via the consent plane; a network mutation is an operate/egress purpose.",
+  "fields": [
+    {"name": "schema", "type": "string", "default": "agent.v1.Action"},
+    {"name": "capability", "type": "string", "doc": "e.g. net.block, net.throttle"},
+    {"name": "args", "type": {"type": "map", "values": "string"}},
+    {"name": "ts", "type": "string"},
+    {"name": "policy", "type": {"type": "map", "values": "string"}, "doc": "space, purpose, gate decision name"}
+  ]
+}
diff --git a/assets/sourceos/schemas/agent/knowledge_update.avsc b/assets/sourceos/schemas/agent/knowledge_update.avsc
new file mode 100644
index 00000000000..40aeb4019b8
--- /dev/null
+++ b/assets/sourceos/schemas/agent/knowledge_update.avsc
@@ -0,0 +1,14 @@
+{
+  "type": "record", "name": "KnowledgeUpdate", "namespace": "agent.v1",
+  "doc": "A System/User Graph delta for hellgraph AtomSpace ingestion.",
+  "fields": [
+    {"name": "schema", "type": "string", "default": "agent.v1.KnowledgeUpdate"},
+    {"name": "graph", "type": {"type": "enum", "name": "Graph", "symbols": ["USER", "SYSTEM"]}},
+    {"name": "ts", "type": "string"},
+    {"name": "patch", "type": {"type": "record", "name": "GraphPatch", "fields": [
+      {"name": "nodes", "type": {"type": "array", "items": {"type": "map", "values": ["null","string","boolean"]}}},
+      {"name": "edges", "type": {"type": "array", "items": {"type": "map", "values": "string"}}}
+    ]}},
+    {"name": "prov", "type": {"type": "map", "values": "string"}}
+  ]
+}
diff --git a/assets/sourceos/schemas/agent/observation.avsc b/assets/sourceos/schemas/agent/observation.avsc
new file mode 100644
index 00000000000..496347d353d
--- /dev/null
+++ b/assets/sourceos/schemas/agent/observation.avsc
@@ -0,0 +1,13 @@
+{
+  "type": "record", "name": "Observation", "namespace": "agent.v1",
+  "doc": "An OS observation emitted by a node agent (e.g. netwatch net.conn / net.beaconing).",
+  "fields": [
+    {"name": "schema", "type": "string", "default": "agent.v1.Observation"},
+    {"name": "source", "type": "string", "doc": "emitting agent, e.g. netwatch"},
+    {"name": "type", "type": "string", "doc": "e.g. net.conn, net.beaconing, net.egress_fanout"},
+    {"name": "ts", "type": "string", "doc": "RFC3339 UTC"},
+    {"name": "attrs", "type": {"type": "map", "values": "string"}},
+    {"name": "severity", "type": {"type": "enum", "name": "Severity", "symbols": ["INFO", "WARN", "CRIT"]}},
+    {"name": "message", "type": ["null", "string"], "default": null}
+  ]
+}
diff --git a/assets/sourceos/tests/test_turtle_netwatch.py b/assets/sourceos/tests/test_turtle_netwatch.py
new file mode 100644
index 00000000000..dfccfca5dc2
--- /dev/null
+++ b/assets/sourceos/tests/test_turtle_netwatch.py
@@ -0,0 +1,102 @@
+"""turtle-netwatch — Network/Connections agent. Proven on synthetic fixtures so
+it runs offline in CI: graph projection, anomaly detection (beaconing +
+egress fan-out), and the fail-closed consent gate on a proposed network action.
+"""
+from __future__ import annotations
+
+import datetime as dt
+import importlib.util
+import json
+import os
+import subprocess
+import sys
+from pathlib import Path
+
+BIN = Path(__file__).resolve().parents[1] / "bin" / "turtle-netwatch"
+
+
+def _load():
+    from importlib.machinery import SourceFileLoader
+    loader = SourceFileLoader("turtle_netwatch", str(BIN))  # extensionless binary
+    spec = importlib.util.spec_from_loader("turtle_netwatch", loader)
+    mod = importlib.util.module_from_spec(spec)
+    loader.exec_module(mod)
+    return mod
+
+
+nw = _load()
+
+
+def _obs(raddr, rport, proc, ts, external="true", sev="INFO"):
+    return {"schema": "agent.v1.Observation", "source": "netwatch", "type": "net.conn",
+            "ts": ts, "severity": sev,
+            "attrs": {"proto": "tcp", "raddr": raddr, "rport": rport, "process": proc,
+                      "pid": "100", "state": "ESTAB", "external": external}}
+
+
+def test_schemas_are_valid_avro_json():
+    d = BIN.parents[1] / "schemas" / "agent"
+    for f in ("observation.avsc", "action.avsc", "knowledge_update.avsc"):
+        j = json.loads((d / f).read_text())
+        assert j["namespace"] == "agent.v1" and j["type"] == "record"
+
+
+def test_graph_projects_processes_hosts_ports():
+    obs = [_obs("93.184.216.34", "443", "curl", "2026-08-02T00:00:00Z")]
+    ku = nw.build_system_graph(obs)
+    kinds = {n["kind"] for n in ku["patch"]["nodes"]}
+    assert {"Process", "Host", "Port"} <= kinds
+    assert ku["graph"] == "SYSTEM"
+    assert ku["patch"]["edges"][0]["rel"] == "CONNECTS_TO"
+
+
+def test_detect_flags_beaconing():
+    # 5 contacts to one dst every 60s, near-zero jitter -> beaconing (CRIT)
+    base = dt.datetime(2026, 8, 2, tzinfo=dt.timezone.utc)
+    obs = [_obs("185.220.101.1", "443", "backdoor",
+                (base + dt.timedelta(seconds=60 * i)).isoformat().replace("+00:00", "Z"))
+           for i in range(5)]
+    findings = nw.detect_anomalies(obs)
+    assert any(f["type"] == "net.beaconing" and f["severity"] == "CRIT" for f in findings)
+
+
+def test_detect_flags_egress_fanout():
+    obs = [_obs(f"203.0.113.{i}", "443", "scanner", "2026-08-02T00:00:00Z") for i in range(12)]
+    findings = nw.detect_anomalies(obs)
+    assert any(f["type"] == "net.egress_fanout" for f in findings)
+
+
+def test_detect_quiet_on_normal_traffic():
+    obs = [_obs("93.184.216.34", "443", "browser", "2026-08-02T00:00:00Z")]
+    assert nw.detect_anomalies(obs) == []
+
+
+def test_propose_fails_closed_without_consent_engine(tmp_path):
+    # no policy-fabric reachable -> a network mutation must be REFUSED (exit 3)
+    env = dict(os.environ)
+    env["SOURCEOS_TERMINAL_RECEIPTS"] = str(tmp_path / "r")
+    env["PROPHET_POLICY_FABRIC"] = str(tmp_path / "nonexistent")
+    r = subprocess.run([sys.executable, str(BIN), "propose", "--action", "block-domain",
+                        "--target", "evil.example", "--json"],
+                       env=env, text=True, capture_output=True)
+    assert r.returncode == 3, r.stderr
+    out = json.loads(r.stdout)
+    assert out["decision"] == "deny"
+    assert any("fail-closed" in x for x in out["denyReasons"])
+    # and it left an auditable refusal receipt
+    receipts = (tmp_path / "netwatch" / "actions.jsonl").read_text()
+    assert "netwatch.action.refused" in receipts
+
+
+def test_snapshot_runs_and_is_shaped(tmp_path):
+    env = dict(os.environ)
+    env["SOURCEOS_TERMINAL_RECEIPTS"] = str(tmp_path / "r")
+    r = subprocess.run([sys.executable, str(BIN), "snapshot", "--json"],
+                       env=env, text=True, capture_output=True)
+    assert r.returncode == 0
+    json.loads(r.stdout)  # a list (possibly empty if no ss/lsof) — must be valid JSON
+
+
+if __name__ == "__main__":
+    import pytest
+    sys.exit(pytest.main([__file__, "-q"]))
diff --git a/packaging/chocolatey/turtleterm/tools/chocolateyInstall.ps1 b/packaging/chocolatey/turtleterm/tools/chocolateyInstall.ps1
index a3d0b747417..1d3819a597b 100644
--- a/packaging/chocolatey/turtleterm/tools/chocolateyInstall.ps1
+++ b/packaging/chocolatey/turtleterm/tools/chocolateyInstall.ps1
@@ -32,7 +32,7 @@ $scripts = @(
   'bin/turtle-copilot', 'bin/turtle-gh', 'bin/turtle-env',
   'bin/turtle-diagnose', 'bin/turtle-apply', 'bin/turtle-chain',
   'bin/turtle-gitea', 'bin/turtle-ci', 'bin/turtle-review',
-  'bin/turtle-watch', 'bin/turtle-cost', 'bin/turtle-bg',
+  'bin/turtle-watch', 'bin/turtle-netwatch', 'bin/turtle-cost', 'bin/turtle-bg',
   'bin/turtle-dash', 'bin/turtle-pr', 'bin/turtle-issue',
   'bin/turtle-hooks', 'bin/turtle-perf', 'bin/turtle-persona',
   'bin/turtle-files', 'bin/turtle-runbook', 'bin/turtle-session',
@@ -89,7 +89,7 @@ New-Item -ItemType Directory -Force -Path $shimDir | Out-Null
 $agentScripts = @(
   'turtle-agentd', 'turtle-agentctl', 'turtle-copilot', 'turtle-gh',
   'turtle-env', 'turtle-diagnose', 'turtle-apply', 'turtle-chain',
-  'turtle-gitea', 'turtle-ci', 'turtle-review', 'turtle-watch',
+  'turtle-gitea', 'turtle-ci', 'turtle-review', 'turtle-watch', 'turtle-netwatch',
   'turtle-cost', 'turtle-bg', 'turtle-dash', 'turtle-pr',
   'turtle-issue', 'turtle-hooks', 'turtle-perf', 'turtle-persona',
   'turtle-files', 'turtle-runbook', 'turtle-session', 'turtle-sync',
diff --git a/packaging/linux/arch/PKGBUILD b/packaging/linux/arch/PKGBUILD
index 3afb19af5a5..a1f9adf237f 100644
--- a/packaging/linux/arch/PKGBUILD
+++ b/packaging/linux/arch/PKGBUILD
@@ -74,7 +74,7 @@ EOF
     turtle-language turtle-session turtle-synapseiq synapseiq-lsp \
     turtle-plan-view turtle-selftest turtle-runbook turtle-voice turtle-sync \
     turtle-perf turtle-persona turtle-files turtle-bg turtle-dash turtle-pr \
-    turtle-issue turtle-hooks turtle-gitea turtle-ci turtle-review turtle-watch \
+    turtle-issue turtle-hooks turtle-gitea turtle-ci turtle-review turtle-watch turtle-netwatch \
     turtle-cost turtle-copilot turtle-gh turtle-env turtle-diagnose \
     turtle-apply turtle-chain turtle-ai-chat; do
     src="assets/sourceos/bin/${script}"
diff --git a/packaging/linux/flatpak/ai.sourceos.TurtleTerm.json b/packaging/linux/flatpak/ai.sourceos.TurtleTerm.json
index f43680202b4..ede215b515e 100644
--- a/packaging/linux/flatpak/ai.sourceos.TurtleTerm.json
+++ b/packaging/linux/flatpak/ai.sourceos.TurtleTerm.json
@@ -37,7 +37,7 @@
         "install -Dm755 target/release/wezterm-mux-server /app/lib/turtleterm/wezterm-mux-server",
         "install -Dm755 assets/sourceos/bin/turtleterm /app/lib/turtleterm-bin/turtleterm",
         "install -Dm755 assets/sourceos/bin/turtleterm-mux-server /app/lib/turtleterm-bin/turtleterm-mux-server",
-        "for script in sourceos-term turtle-term turtle-agentd turtle-agentctl turtle-agent-status turtle-tmux turtle-cloudfog turtle-superconscious turtle-agent-machine turtle-language turtle-session turtle-synapseiq synapseiq-lsp turtle-plan-view turtle-selftest turtle-runbook turtle-voice turtle-sync turtle-perf turtle-persona turtle-files turtle-bg turtle-dash turtle-pr turtle-issue turtle-hooks turtle-gitea turtle-ci turtle-review turtle-watch turtle-cost turtle-copilot turtle-gh turtle-env turtle-diagnose turtle-apply turtle-chain turtle-ai-chat; do [ -f assets/sourceos/bin/$script ] && install -Dm755 assets/sourceos/bin/$script /app/bin/$script || true; done",
+        "for script in sourceos-term turtle-term turtle-agentd turtle-agentctl turtle-agent-status turtle-tmux turtle-cloudfog turtle-superconscious turtle-agent-machine turtle-language turtle-session turtle-synapseiq synapseiq-lsp turtle-plan-view turtle-selftest turtle-runbook turtle-voice turtle-sync turtle-perf turtle-persona turtle-files turtle-bg turtle-dash turtle-pr turtle-issue turtle-hooks turtle-gitea turtle-ci turtle-review turtle-watch turtle-netwatch turtle-cost turtle-copilot turtle-gh turtle-env turtle-diagnose turtle-apply turtle-chain turtle-ai-chat; do [ -f assets/sourceos/bin/$script ] && install -Dm755 assets/sourceos/bin/$script /app/bin/$script || true; done",
         "[ -f assets/sourceos/mcp/turtle-mcp-server ] && install -Dm755 assets/sourceos/mcp/turtle-mcp-server /app/bin/turtle-mcp-server || true",
         "install -d /app/share/turtleterm/shell",
         "for f in assets/sourceos/shell/*; do [ -f $f ] && install -Dm644 $f /app/share/turtleterm/shell/ || true; done",
diff --git a/packaging/linux/rpm/turtle-term.spec b/packaging/linux/rpm/turtle-term.spec
index 955c8d25205..fb36e9c4c0f 100644
--- a/packaging/linux/rpm/turtle-term.spec
+++ b/packaging/linux/rpm/turtle-term.spec
@@ -68,7 +68,7 @@ for script in \
   turtle-language turtle-session turtle-synapseiq synapseiq-lsp \
   turtle-plan-view turtle-selftest turtle-runbook turtle-voice turtle-sync \
   turtle-perf turtle-persona turtle-files turtle-bg turtle-dash turtle-pr \
-  turtle-issue turtle-hooks turtle-gitea turtle-ci turtle-review turtle-watch \
+  turtle-issue turtle-hooks turtle-gitea turtle-ci turtle-review turtle-watch turtle-netwatch \
   turtle-cost turtle-copilot turtle-gh turtle-env turtle-diagnose \
   turtle-apply turtle-chain turtle-ai-chat; do
     src="assets/sourceos/bin/${script}"
diff --git a/packaging/scripts/stage-linux-package.sh b/packaging/scripts/stage-linux-package.sh
index 85f30af1a3d..fc306d0a696 100755
--- a/packaging/scripts/stage-linux-package.sh
+++ b/packaging/scripts/stage-linux-package.sh
@@ -54,7 +54,7 @@ for script in \
   turtle-gitea \
   turtle-ci \
   turtle-review \
-  turtle-watch \
+  turtle-watch turtle-netwatch \
   turtle-cost \
   turtle-copilot \
   turtle-gh \