feat(agentctl): governed web acquisition as first-class acquire command

assets/sourceos/bin/turtle-agentctl

@@ -5,6 +5,7 @@ New verbs (Track A + B):
   ingest-event <json>   Feed a raw terminal event (shell hooks use this)
   noetica-status        Check Noetica reachability and version
   noetica-query <text>  Send a free-text query to Noetica /api/chat
+  acquire --url <URL>   Governed web acquisition via Agent Machine /api/acquire
 """
 
 from __future__ import annotations
@@ -143,6 +144,14 @@ def main(argv: list[str]) -> int:
     noetica_query_p = sub.add_parser("noetica-query", help="send a query to Noetica /api/chat")
     noetica_query_p.add_argument("text", nargs=argparse.REMAINDER)
 
+    # Governed web acquisition via Agent Machine POST /api/acquire (fail-closed).
+    acquire_p = sub.add_parser("acquire", help="governed web acquisition via Agent Machine /api/acquire")
+    acquire_p.add_argument("--url", required=True, help="URL to acquire (required)")
+    acquire_p.add_argument("--account", dest="account", default="", help="account class for the governed run")
+    acquire_p.add_argument("--tier", dest="tier", default="", help="acquisition tier")
+    acquire_p.add_argument("--seeds", dest="seeds", action="append", default=[], help="seed URL/term (repeatable, or comma-separated)")
+    acquire_p.add_argument("--enrich", dest="enrich", action="store_true", help="request downstream enrichment")
+
     # Track D: Policy Fabric
     sub.add_parser("policy-status", help="check Policy Fabric reachability and mode")
 
@@ -516,6 +525,28 @@ def main(argv: list[str]) -> int:
         request = {"action": "noetica_status"}
     elif args.command == "noetica-query":
         request = {"action": "noetica_query", "text": join_remainder(args.text)}
+    elif args.command == "acquire":
+        url = (getattr(args, "url", "") or "").strip()
+        if not url:
+            print(json.dumps({"status": "error", "message": "acquire requires --url"}))
+            return 2
+        # Seeds may be repeated (--seeds a --seeds b) and/or comma-separated (--seeds a,b).
+        seeds: list[str] = []
+        for chunk in getattr(args, "seeds", []) or []:
+            for part in str(chunk).split(","):
+                part = part.strip()
+                if part:
+                    seeds.append(part)
+        request = {
+            "action": "acquire",
+            "url": url,
+            "account_class": getattr(args, "account", "") or "",
+            "tier": getattr(args, "tier", "") or "",
+        }
+        if seeds:
+            request["seeds"] = seeds
+        if getattr(args, "enrich", False):
+            request["enrich"] = True
     elif args.command == "policy-status":
         request = {"action": "policy_status"}
     elif args.command == "bearbrowser-handoff":

assets/sourceos/bin/turtle-agentd

@@ -2329,6 +2329,57 @@ def _handle_request_inner(request: dict[str, Any]) -> dict[str, Any]:  # noqa: C
             ),
         })
 
+    if action == "acquire":
+        url = (request.get("url") or "").strip()
+        if not url:
+            return response("error", {"message": "acquire requires 'url' field"}, status="error")
+        base = agent_machine_url()
+        session_id = env("SOURCEOS_TERMINAL_SESSION_ID") or str(uuid.uuid4())
+        # Governed acquisition request body for Agent Machine POST /api/acquire.
+        payload: dict[str, Any] = {
+            "url": url,
+            "accountClass": request.get("account_class", "") or "",
+            "tier": request.get("tier", "") or "",
+        }
+        seeds = request.get("seeds")
+        if seeds:
+            payload["seeds"] = seeds
+        if request.get("enrich"):
+            payload["enrich"] = True
+        result = _http_post(base.rstrip("/") + "/api/acquire", payload, timeout=15)
+        # Fail closed: if Agent Machine is unreachable, the acquisition is NOT admitted.
+        reachable = result is not None
+        status_label = "dispatched" if reachable else "blocked_offline"
+        receipt = write_receipt({
+            "schema": "sourceos.terminal.event.v0",
+            "event_type": "agent-machine.acquire",
+            "session_id": session_id,
+            "url": url,
+            "account_class": payload["accountClass"],
+            "tier": payload["tier"],
+            "agent_machine_url": base,
+            "reachable": reachable,
+            "status": status_label,
+            "capturedAt": utc_now(),
+        })
+        return response("acquisition_result", {
+            "url": url,
+            "account_class": payload["accountClass"],
+            "tier": payload["tier"],
+            "seeds": payload.get("seeds", []),
+            "enrich": bool(payload.get("enrich", False)),
+            "agent_machine_url": base,
+            "reachable": reachable,
+            "status": status_label,
+            "acquisition_response": result,
+            "receipt": receipt,
+            "decision": decision_stub(
+                "agent-machine.acquire",
+                f"Governed web acquisition {'dispatched to' if reachable else 'blocked (Agent Machine unreachable) at'} {base}/api/acquire.",
+                "allow" if reachable else "deny",
+            ),
+        }, status="ok" if reachable else "error")
+
     # ------------------------------------------------------------------
     # CloudFog
     # ------------------------------------------------------------------

assets/sourceos/skills/turtle-agent-machine-bridge.json

@@ -12,7 +12,8 @@
       "agent-machine.probe",
       "agent-machine.render",
       "agent-machine.evaluate",
-      "agent-machine.receipts"
+      "agent-machine.receipts",
+      "agent-machine.acquire"
     ],
     "filePatterns": [
       "*.agent-pod.json"
@@ -22,7 +23,8 @@
       "agentpod",
       "runtime-surface",
       "activation-decision",
-      "runtime-evidence"
+      "runtime-evidence",
+      "web-acquisition"
     ]
   },
   "requires": {
@@ -47,7 +49,8 @@
     "agentpod-plan",
     "activation-decision",
     "runtime-evidence-reference",
-    "storage-receipt-reference"
+    "storage-receipt-reference",
+    "acquisition-provenance-receipt"
   ],
   "reviewMode": true,
   "allowShellExecution": false,

assets/sourceos/tests/test_turtle_agentd.py

@@ -75,6 +75,28 @@ def main() -> int:
         assert summary["kind"] == "summary"
         assert summary["data"]["event_count"] == 0
 
+        # Governed web acquisition: turtle-agentctl builds the request dict and
+        # turtle-agentd routes it to Agent Machine /api/acquire, failing closed
+        # when Agent Machine is unreachable.
+        acq_env = dict(env)
+        acq_env["SOURCEOS_AGENT_MACHINE_URL"] = "http://127.0.0.1:9"  # discard/refused -> offline
+        acquire = run_agentctl(
+            ["acquire", "--url", "https://example.com/data",
+             "--account", "research", "--tier", "gold",
+             "--seeds", "https://a.example,https://b.example", "--enrich"],
+            acq_env,
+        )
+        assert acquire["kind"] == "acquisition_result"
+        assert acquire["status"] == "error"  # fail-closed when unreachable
+        assert acquire["data"]["url"] == "https://example.com/data"
+        assert acquire["data"]["account_class"] == "research"
+        assert acquire["data"]["tier"] == "gold"
+        assert acquire["data"]["seeds"] == ["https://a.example", "https://b.example"]
+        assert acquire["data"]["enrich"] is True
+        assert acquire["data"]["reachable"] is False
+        assert acquire["data"]["status"] == "blocked_offline"
+        assert acquire["data"]["decision"]["decision"] == "deny"
+
     return 0
 
 
@@ -132,3 +154,71 @@ def test_policy_evaluate_denies_agent_egress():
                             execution_domain="host", actor_id="agent:autonomous")
     assert r["decision"]["outcome"] == "deny"
     assert r["source"] == "consent-plane"
+
+
+# --------------------------------------------------------------------- acquire
+def test_acquire_action_posts_to_api_acquire(tmp_path, monkeypatch):
+    """The `acquire` action POSTs the governed body to Agent Machine /api/acquire."""
+    monkeypatch.setenv("SOURCEOS_TERMINAL_RECEIPTS", str(tmp_path / "receipts"))
+    monkeypatch.setenv("SOURCEOS_AGENT_MACHINE_URL", "http://agent-machine.local:9000")
+
+    captured: dict = {}
+
+    def fake_post(url, payload, timeout=5):
+        captured["url"] = url
+        captured["payload"] = payload
+        captured["timeout"] = timeout
+        return {"runId": "urn:srcos:acquire:run:test", "admitted": True}
+
+    monkeypatch.setattr(_ad, "_http_post", fake_post)
+
+    resp = _ad.handle_request({
+        "action": "acquire",
+        "url": "https://example.com/report",
+        "account_class": "research",
+        "tier": "gold",
+        "seeds": ["https://seed.example"],
+        "enrich": True,
+    })
+
+    # Routed to the correct Agent Machine endpoint with the governed body shape.
+    assert captured["url"] == "http://agent-machine.local:9000/api/acquire"
+    assert captured["payload"] == {
+        "url": "https://example.com/report",
+        "accountClass": "research",
+        "tier": "gold",
+        "seeds": ["https://seed.example"],
+        "enrich": True,
+    }
+    assert resp["status"] == "ok"
+    assert resp["kind"] == "acquisition_result"
+    assert resp["data"]["reachable"] is True
+    assert resp["data"]["status"] == "dispatched"
+    assert resp["data"]["acquisition_response"] == {"runId": "urn:srcos:acquire:run:test", "admitted": True}
+    assert resp["data"]["decision"]["decision"] == "allow"
+    assert resp["data"]["decision"]["action"] == "agent-machine.acquire"
+
+
+def test_acquire_fails_closed_when_agent_machine_unreachable(tmp_path, monkeypatch):
+    """Fail-closed: unreachable Agent Machine yields deny + error, no admission."""
+    monkeypatch.setenv("SOURCEOS_TERMINAL_RECEIPTS", str(tmp_path / "receipts"))
+    monkeypatch.setattr(_ad, "_http_post", lambda url, payload, timeout=5: None)
+
+    resp = _ad.handle_request({
+        "action": "acquire",
+        "url": "https://example.com/report",
+        "account_class": "research",
+        "tier": "gold",
+    })
+    assert resp["status"] == "error"
+    assert resp["data"]["reachable"] is False
+    assert resp["data"]["status"] == "blocked_offline"
+    assert resp["data"]["acquisition_response"] is None
+    assert resp["data"]["decision"]["decision"] == "deny"
+
+
+def test_acquire_requires_url(tmp_path, monkeypatch):
+    monkeypatch.setenv("SOURCEOS_TERMINAL_RECEIPTS", str(tmp_path / "receipts"))
+    resp = _ad.handle_request({"action": "acquire", "url": ""})
+    assert resp["status"] == "error"
+    assert "url" in resp["data"]["message"]