Add policy engine example scripts for agent permissions testing [CC-18, CC-19, CC-20, CC-21]

Makefile

@@ -26,7 +26,11 @@ clean: ## Removes all generated code (except _patch.py files)
 	@printf "=== Cleaning src directory\n"
 	@rm -rf src/pydo/resources
 	@rm -rf src/pydo/types
-	@find src/pydo -type f ! -name "_patch.py" ! -name "custom_*.py" ! -name "exceptions.py" -exec rm -rf {} +
+	@find src/pydo -type f \
+		! -name "_patch.py" ! -name "custom_*.py" ! -name "exceptions.py" \
+		! -path "*/agents/__init__.py" ! -path "*/aio/agents/__init__.py" \
+		! -path "*/agents/session.py" ! -path "*/aio/agents/session.py" \
+		-exec rm -rf {} +
 
 .PHONY: download-spec
 download-spec: ## Download Latest DO Spec

examples/agents/async_stream_session.py

@@ -0,0 +1,24 @@
+"""Async stream session events. Set DIGITALOCEAN_TOKEN and SESSION_ID."""
+
+import asyncio
+import os
+
+from pydo.aio import Client
+
+SESSION_ID = os.environ["SESSION_ID"]
+
+
+async def main() -> None:
+    async with Client(token=os.environ["DIGITALOCEAN_TOKEN"]) as client:
+        stream = await client.agents.sessions.stream(SESSION_ID)
+        async with stream as events:
+            async for event in events:
+                if getattr(event, "type", None) == "run.token_delta" and event.get("data"):
+                    print(event.data.text, end="", flush=True)
+                elif "token_chunk" in event:
+                    print(event.token_chunk.text, end="", flush=True)
+                else:
+                    print(event)
+
+
+asyncio.run(main())

examples/agents/attach.py

@@ -0,0 +1,53 @@
+"""Attach to an EXISTING agent session and stream one turn.
+
+Like ``doctl agents attach``: connect to a session that already exists, send a
+prompt, and consume the SSE event feed through the high-level API — no thread,
+no raw event-string matching. attach() never destroys the session, so it stays
+alive for further turns.
+
+Get a SESSION_ID first by creating one (the session stays up):
+    AGENT_SPEC=... python examples/agents/create_session.py
+then copy the "session_id" it prints.
+
+Required env:
+  DIGITALOCEAN_TOKEN
+  PYDO_AGENTS_ENDPOINT   stage2: https://api.s2r1.internal.digitalocean.com
+  SESSION_ID             the session to attach to
+
+Optional env:
+  PROMPT                 message to send (default below)
+"""
+
+import os
+import sys
+
+from pydo import Client
+from pydo.agents import AgentEventType
+
+client = Client(
+    token=os.environ["DIGITALOCEAN_TOKEN"],
+    agents_endpoint=os.environ.get("PYDO_AGENTS_ENDPOINT"),
+)
+
+agent = client.agents.attach(os.environ["SESSION_ID"])
+prompt = os.environ.get("PROMPT", "In one short sentence, what is DigitalOcean?")
+
+print(f"[attached {agent.session_id}]\n>>> {prompt}\n", file=sys.stderr)
+
+stream = agent.run_streamed(prompt)  # opens the stream, sends input, yields events
+for event in stream:
+    if event.type == AgentEventType.TOKEN:
+        print(event.text, end="", flush=True)  # live reply -> stdout
+    elif event.type == AgentEventType.TOOL_CALL:
+        print(f"\n[tool] {event.tool_name}", file=sys.stderr)
+    elif event.type == AgentEventType.HITL_REQUESTED:
+        print(f"\n[hitl auto-approved] {event.request_id}", file=sys.stderr)
+
+result = stream.result
+print(
+    f"\n\n[{result.status}] captured {len(result.final_output)} chars "
+    f"(tokens out={result.usage.get('tokens_out')})",
+    file=sys.stderr,
+)
+# attach() leaves the session running; destroy it when done:
+#   SESSION_ID=... python examples/agents/destroy_session.py

examples/agents/attach_by_name.py

@@ -0,0 +1,52 @@
+"""Look up a hosted-agent session by name (instead of by id).
+
+Sessions can be filtered server-side by name (``GET /v2/agents/sessions?name=``).
+This script lists the matches and resolves the name to a session handle via
+``client.agents.attach_by_name(...)``.
+
+Required env:
+  DIGITALOCEAN_TOKEN
+  PYDO_AGENTS_ENDPOINT   stage2: https://api.s2r1.internal.digitalocean.com
+  SESSION_NAME           the session name to look up
+"""
+
+import os
+import sys
+
+from pydo import Client
+
+
+def main() -> int:
+    name = os.environ["SESSION_NAME"]
+
+    client = Client(
+        token=os.environ["DIGITALOCEAN_TOKEN"],
+        agents_endpoint=os.environ.get("PYDO_AGENTS_ENDPOINT"),
+    )
+
+    resp = client.agents.sessions.list(name=name)
+    sessions = resp.get("sessions", []) if hasattr(resp, "get") else []
+    print(f"[list name={name!r}] {len(sessions)} match(es):", file=sys.stderr)
+    for s in sessions:
+        print(f"  {s['session_id']}  {s['name']}  {s['status']}", file=sys.stderr)
+
+    if not sessions:
+        # Help diagnose a 0-match result: show what names actually exist now.
+        all_resp = client.agents.sessions.list()
+        all_sessions = all_resp.get("sessions", []) if hasattr(all_resp, "get") else []
+        print(
+            f"[attach_by_name] no session named {name!r}. "
+            f"{len(all_sessions)} session(s) currently exist:",
+            file=sys.stderr,
+        )
+        for s in all_sessions:
+            print(f"  {s['name']}  ({s['status']})", file=sys.stderr)
+        return 1
+
+    agent = client.agents.attach_by_name(name)
+    print(f"[attach_by_name] resolved session_id: {agent.session_id}", file=sys.stderr)
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

examples/agents/create_session.py

@@ -0,0 +1,26 @@
+"""Create a session from an agent manifest (``agents.yaml``).
+
+A session is created entirely from the agent spec — the runtime adapter,
+sandbox template, env vars, etc. are all defined in the manifest. The client
+uploads it verbatim (``Content-Type: application/x-yaml``); the server parses
+and validates it.
+
+Required env:
+  DIGITALOCEAN_TOKEN
+  PYDO_AGENTS_ENDPOINT  (stage2: https://api.s2r1.internal.digitalocean.com)
+  AGENT_SPEC            path to the agent spec YAML (default: agent-spec.yaml)
+"""
+
+import json
+import os
+
+from pydo import Client
+
+spec_path = os.environ.get("AGENT_SPEC", "agent-spec.yaml")
+with open(spec_path, "r", encoding="utf-8") as fh:
+    manifest = fh.read()
+
+client = Client(token=os.environ["DIGITALOCEAN_TOKEN"])
+resp = client.agents.sessions.create_from_manifest(manifest)
+
+print(json.dumps(resp, indent=2, default=str))

examples/agents/destroy_session.py

@@ -0,0 +1,12 @@
+"""Destroy a session. Set DIGITALOCEAN_TOKEN and SESSION_ID."""
+
+import os
+
+from pydo import Client
+
+SESSION_ID = os.environ["SESSION_ID"]
+
+client = Client(token=os.environ["DIGITALOCEAN_TOKEN"])
+client.agents.sessions.destroy(SESSION_ID)
+
+print("ok", SESSION_ID)

examples/agents/get_session.py

@@ -0,0 +1,13 @@
+"""Get a session. Set DIGITALOCEAN_TOKEN and SESSION_ID."""
+
+import json
+import os
+
+from pydo import Client
+
+SESSION_ID = os.environ["SESSION_ID"]
+
+client = Client(token=os.environ["DIGITALOCEAN_TOKEN"])
+resp = client.agents.sessions.get(SESSION_ID)
+
+print(json.dumps(resp, indent=2, default=str))

examples/agents/list_sessions.py

@@ -0,0 +1,11 @@
+"""List sessions. Set DIGITALOCEAN_TOKEN (and PYDO_AGENTS_ENDPOINT for stage2)."""
+
+import os
+
+from pydo import Client
+
+client = Client(token=os.environ["DIGITALOCEAN_TOKEN"])
+
+resp = client.agents.sessions.list(page_size=50)
+for session in resp.get("sessions", []):
+    print(session.session_id, session.status, session.agent_kind)

examples/agents/pause_resume.py

@@ -0,0 +1,76 @@
+"""Pause and resume a hosted-agent session.
+
+Attaches to an existing session (by id or by name), pauses it, waits for it to
+report ``SESSION_STATUS_PAUSED``, then resumes it and waits for ``READY``.
+
+Required env:
+  DIGITALOCEAN_TOKEN
+  PYDO_AGENTS_ENDPOINT   stage2: https://api.s2r1.internal.digitalocean.com
+
+Pick a session with one of (SESSION_ID takes precedence):
+  SESSION_ID             the session id to pause/resume
+  SESSION_NAME           resolve the session by name instead
+"""
+
+import os
+import sys
+import time
+
+from pydo import Client
+from pydo.agents.custom_models import SessionStatus
+
+
+def _wait_for_status(agent, target, *, timeout=120.0, poll_interval=2.0):
+    """Poll until the session reaches ``target`` (or a terminal/failed state)."""
+    deadline = time.monotonic() + timeout
+    while True:
+        agent.refresh()
+        status = agent.status
+        print(f"  status: {status}", file=sys.stderr)
+        if status == target:
+            return
+        if status in (SessionStatus.FAILED, SessionStatus.DESTROYED):
+            raise RuntimeError(f"session {agent.session_id} is {status}")
+        if time.monotonic() > deadline:
+            raise TimeoutError(
+                f"session {agent.session_id} did not reach {target} in {timeout}s "
+                f"(last status: {status})"
+            )
+        time.sleep(poll_interval)
+
+
+def main() -> int:
+    client = Client(
+        token=os.environ["DIGITALOCEAN_TOKEN"],
+        agents_endpoint=os.environ.get("PYDO_AGENTS_ENDPOINT"),
+    )
+
+    session_id = os.environ.get("SESSION_ID")
+    session_name = os.environ.get("SESSION_NAME")
+    if session_id:
+        agent = client.agents.attach(session_id)
+    elif session_name:
+        agent = client.agents.attach_by_name(session_name)
+    else:
+        print("set SESSION_ID or SESSION_NAME", file=sys.stderr)
+        return 2
+
+    print(f"[attach] session_id={agent.session_id}", file=sys.stderr)
+    agent.refresh()
+    print(f"[attach] current status: {agent.status}", file=sys.stderr)
+
+    print("[pause] pausing session...", file=sys.stderr)
+    agent.pause()
+    _wait_for_status(agent, SessionStatus.PAUSED)
+    print("[pause] session is PAUSED", file=sys.stderr)
+
+    print("[resume] resuming session...", file=sys.stderr)
+    agent.resume()
+    _wait_for_status(agent, SessionStatus.READY)
+    print("[resume] session is READY", file=sys.stderr)
+
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

examples/agents/policy_auto_allow.py

@@ -0,0 +1,99 @@
+"""Policy engine test: auto-allow rule.
+
+Creates a session with ``defaultAction: ask`` and a single ``touch * → allow``
+rule, then asks the agent to run ``touch /workspace/allow_test.txt``.
+
+Expected: the Bash tool call matches the allow rule and executes WITHOUT any
+HITL prompt being raised. The test FAILS if a ``hitl_requested`` event fires.
+
+Required env:
+  DIGITALOCEAN_TOKEN
+  AGENT_SPEC   path to the base agents.yaml (default: agent-spec.yaml)
+
+Optional env:
+  PYDO_AGENTS_ENDPOINT   stage2: https://api.s2r1.internal.digitalocean.com
+"""
+
+import os
+import sys
+
+import yaml
+
+from pydo import Client
+from pydo.agents import AgentEventType
+
+
+_PERMISSIONS = {
+    "defaultAction": "ask",
+    "rules": [
+        {"tool": "bash", "match": {"command": "touch *"}, "action": "allow"},
+        {"tool": "bash", "match": {"command": "git status"}, "action": "allow"},
+    ],
+}
+
+_PROMPT = (
+    "Use bash to run exactly this command and nothing else: "
+    "touch /workspace/allow_test.txt"
+)
+
+
+def _load_manifest(name: str, permissions: dict) -> str:
+    spec_path = os.environ.get("AGENT_SPEC", "agent-spec.yaml")
+    with open(spec_path, "r", encoding="utf-8") as fh:
+        spec = yaml.safe_load(fh)
+    spec["metadata"]["name"] = name
+    spec["spec"]["permissions"] = permissions
+    return yaml.dump(spec, default_flow_style=False)
+
+
+def main() -> int:
+    client = Client(
+        token=os.environ["DIGITALOCEAN_TOKEN"],
+        agents_endpoint=os.environ.get("PYDO_AGENTS_ENDPOINT"),
+    )
+
+    manifest = _load_manifest("policy-auto-allow-test", _PERMISSIONS)
+    hitl_events = []
+
+    def _track_and_approve(event):
+        hitl_events.append(event)
+        return "approve"  # resolve so the run doesn't hang if HITL fires unexpectedly
+
+    with client.agents.start(manifest) as agent:
+        print(f"[session {agent.session_id}]", file=sys.stderr)
+        print(f"[policy] defaultAction=ask  touch * → allow", file=sys.stderr)
+        print(f"[prompt] {_PROMPT}\n", file=sys.stderr)
+
+        stream = agent.run_streamed(_PROMPT, hitl=_track_and_approve, timeout=180)
+        for event in stream:
+            if event.type == AgentEventType.TOKEN:
+                print(event.text, end="", flush=True)
+            elif event.type == AgentEventType.TOOL_CALL:
+                print(f"\n[tool_call] {event.tool_name}", file=sys.stderr)
+            elif event.type == AgentEventType.HITL_REQUESTED:
+                print(f"\n[hitl_fired] request_id={event.request_id}", file=sys.stderr)
+
+        result = stream.result
+        print(f"\n[{result.status}]", file=sys.stderr)
+
+    if hitl_events:
+        print(
+            f"\nFAIL HITL fired {len(hitl_events)} time(s) — "
+            "touch should have been auto-allowed without prompting",
+            file=sys.stderr,
+        )
+        return 1
+
+    if result.status != "completed":
+        print(
+            f"\nFAIL run ended with status={result.status!r}",
+            file=sys.stderr,
+        )
+        return 1
+
+    print("\nPASS touch ran without HITL prompt (auto-allow confirmed)")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())