From 7ff6bf858dfa687c0844c8ae88f6d61f39fca05c Mon Sep 17 00:00:00 2001 From: Roger Oriol Date: Sun, 26 Jul 2026 21:07:47 +0200 Subject: [PATCH] security agent cleanup --- README.md | 8 +- agent-human-in-the-loop/agent.py | 40 +++--- agent-human-in-the-loop/tools/interaction.py | 2 +- agent-human-in-the-loop/tools/registry.py | 2 +- agent-human-in-the-loop/tools/todo.py | 2 +- agent-planning/agent.py | 34 +++--- agent-planning/tools/todo.py | 2 +- agent-security/agent.py | 122 +++++++++---------- agent-security/prompt_safety.py | 2 +- agent-security/resource_limits.py | 48 ++++---- agent-security/secret_management.py | 30 ++--- agent-security/session_control.py | 23 ++-- agent-security/tool_policy.py | 14 +-- agent-security/tools/_dispatch.py | 2 +- agent-security/tools/audit.py | 22 ++-- agent-security/tools/interaction.py | 2 +- agent-security/tools/registry.py | 2 +- agent-security/tools/sandbox.py | 12 +- agent-security/tools/todo.py | 2 +- agent-security/tools/validators.py | 28 +++-- agent-with-tools/agent.py | 2 +- 21 files changed, 206 insertions(+), 195 deletions(-) diff --git a/README.md b/README.md index 78c50c9..fa56fcd 100644 --- a/README.md +++ b/README.md @@ -4,16 +4,16 @@ Companion code for the [ruxu.dev](https://www.ruxu.dev) blog post series on buil ## Blog Post Series -1. [Build a Basic AI Agent](https://www.ruxu.dev/articles/ai/build-a-basic-ai-agent/) — A minimal conversational agent with a message loop, powered by a local model via Ollama. -2. [Build an AI Agent with Tools](https://www.ruxu.dev/articles/ai/build-an-ai-agent-with-tools/) — Extends the agent with a tool registry so the LLM can read/write files, search the filesystem, run shell commands, and fetch web pages. +1. [Build a Basic AI Agent](https://www.ruxu.dev/articles/ai/build-a-basic-ai-agent/): A minimal conversational agent with a message loop, powered by a local model via Ollama. +2. [Build an AI Agent with Tools](https://www.ruxu.dev/articles/ai/build-an-ai-agent-with-tools/): Extends the agent with a tool registry so the LLM can read/write files, search the filesystem, run shell commands, and fetch web pages. ## Structure ``` -simple-agent/ # Part 1 — bare-bones agent loop +simple-agent/ # Part 1: bare-bones agent loop agent.py -agent-with-tools/ # Part 2 — agent with tool-calling support +agent-with-tools/ # Part 2: agent with tool-calling support agent.py tools/ filesystem.py # read, write, search files diff --git a/agent-human-in-the-loop/agent.py b/agent-human-in-the-loop/agent.py index d2da994..16197f1 100644 --- a/agent-human-in-the-loop/agent.py +++ b/agent-human-in-the-loop/agent.py @@ -62,7 +62,7 @@ def _ask_permission(tool_name: str, args: dict) -> bool: try: answer = input(" Allow this action? [y/n]: ").strip().lower() except EOFError: - print(" (EOF — denying permission)") + print(" (EOF, denying permission)") return False if answer in ("y", "yes"): return True @@ -107,7 +107,7 @@ def check_permission( if mode == PermissionMode.ACCEPT_EDITS and tool_name in WRITE_TOOLS: path = _resolve_tool_path(tool_name, args) if path and _is_within_working_dir(path, working_dir): - return True # auto-approved — within the working directory + return True # auto-approved, within the working directory # Path is outside the working directory → fall through to ask # Default mode, or acceptEdits for non-write / out-of-tree tools @@ -186,7 +186,7 @@ def agent_loop(client, mode: PermissionMode, working_dir: Path): "- Clarification (ask_question): ask the user a single focused question when you " "are genuinely blocked and cannot reasonably infer the missing information from " "context. Do not use it for progress updates or to confirm actions you can already " - "take — only ask when it is strictly necessary to proceed.\n\n" + "take, only ask when it is strictly necessary to proceed.\n\n" "## Working directory\n\n" "The current working directory is always the user's project root. " @@ -202,22 +202,22 @@ def agent_loop(client, mode: PermissionMode, working_dir: Path): "todo_append (status: pending).\n" "3. Before starting a step, mark it in_progress with todo_update. " "Keep only one item in_progress at a time.\n" - "4. Mark items done immediately after completing them — do not batch completions.\n" + "4. Mark items done immediately after completing them, do not batch completions.\n" "5. Call todo_list to review remaining work before moving to the next step.\n" "6. Mark tasks cancelled if they become unnecessary.\n\n" "For simple, single-step tasks: act directly without creating todos.\n\n" "Planning tool calls (write_scratchpad, todo_append, todo_update, todo_list) " "are internal bookkeeping, not responses to the user. After any planning tool " - "call, always continue working immediately — make your next tool call or, once " + "call, always continue working immediately, make your next tool call or, once " "the task is fully complete, give a substantive final answer. " "Never emit an empty or whitespace-only message.\n\n" "## Replanning\n\n" "After every tool result, check whether the outcome matched your expectation. " "If a tool returns an error, unexpected output, or reveals information that " - "changes your understanding of the task, do not move to the next planned step — " + "changes your understanding of the task, do not move to the next planned step, " "replan first.\n\n" "When a step fails:\n" - "1. Diagnose in the scratchpad — is this a recoverable input error (wrong path, " + "1. Diagnose in the scratchpad: is this a recoverable input error (wrong path, " "typo, wrong argument) or a deeper problem (wrong approach, wrong assumption)?\n" "2. Mark the task failed: todo_update(id, status='failed').\n" "3. Choose a recovery action:\n" @@ -227,8 +227,8 @@ def agent_loop(client, mode: PermissionMode, working_dir: Path): " - Reorder: new information makes a different task more urgent. Update the " "pending items before continuing.\n" "4. If todo_update reports that the retry limit has been reached, stop retrying. " - "Write a clear diagnosis in the scratchpad — what you tried, what failed each " - "time, and what you need — then give the user a concise escalation message " + "Write a clear diagnosis in the scratchpad: what you tried, what failed each " + "time, and what you need, then give the user a concise escalation message " "and wait for their input.\n\n" "When a tool succeeds but returns information that changes the picture, pause " "before acting. Call todo_list, reassess all pending items in the scratchpad, " @@ -236,34 +236,34 @@ def agent_loop(client, mode: PermissionMode, working_dir: Path): "## How to use the scratchpad\n\n" "Before each tool call during a complex task, update the scratchpad with your " "current thinking. Structure each entry around these five steps:\n\n" - "1. Restate the goal — write what you understand the task to be, in your own words. " + "1. Restate the goal: write what you understand the task to be, in your own words. " "This catches misreads before they compound into wasted work.\n" - "2. Survey what you know — note which files you have seen, what the code structure " + "2. Survey what you know: note which files you have seen, what the code structure " "looks like, and what constraints or requirements apply.\n" - "3. Evaluate options — reason through at least two approaches and explain why you " + "3. Evaluate options: reason through at least two approaches and explain why you " "are choosing one over the other (e.g. 'I could rewrite the middleware, or wrap it. " "Wrapping is safer because it leaves the existing call sites untouched.').\n" - "4. Anticipate failure modes — write down what could go wrong with the chosen " + "4. Anticipate failure modes: write down what could go wrong with the chosen " "approach and how you would diagnose it (e.g. 'If the tests fail after this, the " "most likely cause is that the session cookie name changed.').\n" - "5. Decide the next single action — commit to exactly one tool call. " + "5. Decide the next single action: commit to exactly one tool call. " "Do not plan several calls at once; decide the next step only.\n\n" "Re-read the scratchpad whenever you resume after a tool result to keep your " "reasoning grounded in what you have already learned.\n\n" "## Done detection\n\n" "Do not give a final answer based on the task list being empty alone. " "Before declaring the task complete, verify all three of the following:\n\n" - "1. Structural completion — call todo_list and confirm there are no pending, " + "1. Structural completion: call todo_list and confirm there are no pending, " "in_progress, or failed items.\n" - "2. Verification — check the output against the original goal. For code tasks: " + "2. Verification: check the output against the original goal. For code tasks: " "run the tests or build with run_bash and confirm they pass. For research tasks: " "re-read the scratchpad and confirm the assembled answer addresses what was " "actually asked.\n" - "3. Uncertainty check — read the scratchpad and ask: are there unresolved " + "3. Uncertainty check: read the scratchpad and ask: are there unresolved " "questions, assumptions that were never validated, or tasks that were cancelled " "rather than properly completed?\n\n" "If all three are satisfied, give your final answer. If any are not, re-enter " - "the planning loop — add the outstanding items to the todo list and continue." + "the planning loop, add the outstanding items to the todo list and continue." ), } ] @@ -290,10 +290,10 @@ def agent_loop(client, mode: PermissionMode, working_dir: Path): messages.append(message) if message.tool_calls: - # The LLM wants to use one or more tools — run them, then loop + # The LLM wants to use one or more tools, run them, then loop handle_tool_calls(message.tool_calls, messages, mode, working_dir) elif not message.content or not message.content.strip(): - # The model ended its turn with an empty message — most commonly + # The model ended its turn with an empty message, most commonly # happens after a planning-only tool call (scratchpad / todo). # Nudge it to continue rather than silently stalling. messages.append({ diff --git a/agent-human-in-the-loop/tools/interaction.py b/agent-human-in-the-loop/tools/interaction.py index 34a3540..6af941e 100644 --- a/agent-human-in-the-loop/tools/interaction.py +++ b/agent-human-in-the-loop/tools/interaction.py @@ -4,5 +4,5 @@ def ask_question(question: str) -> str: try: answer = input(" Your answer: ").strip() except EOFError: - return "(no answer — EOF)" + return "(no answer, EOF)" return answer if answer else "(no answer provided)" diff --git a/agent-human-in-the-loop/tools/registry.py b/agent-human-in-the-loop/tools/registry.py index dbf7c40..64197c0 100644 --- a/agent-human-in-the-loop/tools/registry.py +++ b/agent-human-in-the-loop/tools/registry.py @@ -265,7 +265,7 @@ def get_tool_schemas(): "and cannot reasonably infer it from context. " "Ask one focused question at a time. " "Do not use this for progress updates or to confirm actions you can already " - "take — only ask when you are genuinely blocked." + "take, only ask when you are genuinely blocked." ), "parameters": { "type": "object", diff --git a/agent-human-in-the-loop/tools/todo.py b/agent-human-in-the-loop/tools/todo.py index a11a0a9..4fcb6db 100644 --- a/agent-human-in-the-loop/tools/todo.py +++ b/agent-human-in-the-loop/tools/todo.py @@ -102,7 +102,7 @@ def todo_update(id, content=None, status=None) -> str: if item["status"] == "in_progress" and retries > 0: if retries >= RETRY_LIMIT: return ( - f"Updated to do item {id} to in_progress — " + f"Updated to do item {id} to in_progress, " f"but this is retry {retries} of { RETRY_LIMIT} (retry limit reached). " f"Do not retry again. Escalate to the user instead." diff --git a/agent-planning/agent.py b/agent-planning/agent.py index 4d2c18a..013e256 100644 --- a/agent-planning/agent.py +++ b/agent-planning/agent.py @@ -82,22 +82,22 @@ def agent_loop(client): "todo_append (status: pending).\n" "3. Before starting a step, mark it in_progress with todo_update. " "Keep only one item in_progress at a time.\n" - "4. Mark items done immediately after completing them — do not batch completions.\n" + "4. Mark items done immediately after completing them, do not batch completions.\n" "5. Call todo_list to review remaining work before moving to the next step.\n" "6. Mark tasks cancelled if they become unnecessary.\n\n" "For simple, single-step tasks: act directly without creating todos.\n\n" "Planning tool calls (write_scratchpad, todo_append, todo_update, todo_list) " "are internal bookkeeping, not responses to the user. After any planning tool " - "call, always continue working immediately — make your next tool call or, once " + "call, always continue working immediately, make your next tool call or, once " "the task is fully complete, give a substantive final answer. " "Never emit an empty or whitespace-only message.\n\n" "## Replanning\n\n" "After every tool result, check whether the outcome matched your expectation. " "If a tool returns an error, unexpected output, or reveals information that " - "changes your understanding of the task, do not move to the next planned step — " + "changes your understanding of the task, do not move to the next planned step, " "replan first.\n\n" "When a step fails:\n" - "1. Diagnose in the scratchpad — is this a recoverable input error (wrong path, " + "1. Diagnose in the scratchpad: is this a recoverable input error (wrong path, " "typo, wrong argument) or a deeper problem (wrong approach, wrong assumption)?\n" "2. Mark the task failed: todo_update(id, status='failed').\n" "3. Choose a recovery action:\n" @@ -107,8 +107,8 @@ def agent_loop(client): " - Reorder: new information makes a different task more urgent. Update the " "pending items before continuing.\n" "4. If todo_update reports that the retry limit has been reached, stop retrying. " - "Write a clear diagnosis in the scratchpad — what you tried, what failed each " - "time, and what you need — then give the user a concise escalation message " + "Write a clear diagnosis in the scratchpad: what you tried, what failed each " + "time, and what you need, then give the user a concise escalation message " "and wait for their input.\n\n" "When a tool succeeds but returns information that changes the picture, pause " "before acting. Call todo_list, reassess all pending items in the scratchpad, " @@ -116,34 +116,34 @@ def agent_loop(client): "## How to use the scratchpad\n\n" "Before each tool call during a complex task, update the scratchpad with your " "current thinking. Structure each entry around these five steps:\n\n" - "1. Restate the goal — write what you understand the task to be, in your own words. " + "1. Restate the goal: write what you understand the task to be, in your own words. " "This catches misreads before they compound into wasted work.\n" - "2. Survey what you know — note which files you have seen, what the code structure " + "2. Survey what you know: note which files you have seen, what the code structure " "looks like, and what constraints or requirements apply.\n" - "3. Evaluate options — reason through at least two approaches and explain why you " + "3. Evaluate options: reason through at least two approaches and explain why you " "are choosing one over the other (e.g. 'I could rewrite the middleware, or wrap it. " "Wrapping is safer because it leaves the existing call sites untouched.').\n" - "4. Anticipate failure modes — write down what could go wrong with the chosen " + "4. Anticipate failure modes: write down what could go wrong with the chosen " "approach and how you would diagnose it (e.g. 'If the tests fail after this, the " "most likely cause is that the session cookie name changed.').\n" - "5. Decide the next single action — commit to exactly one tool call. " + "5. Decide the next single action: commit to exactly one tool call. " "Do not plan several calls at once; decide the next step only.\n\n" "Re-read the scratchpad whenever you resume after a tool result to keep your " "reasoning grounded in what you have already learned.\n\n" "## Done detection\n\n" "Do not give a final answer based on the task list being empty alone. " "Before declaring the task complete, verify all three of the following:\n\n" - "1. Structural completion — call todo_list and confirm there are no pending, " + "1. Structural completion: call todo_list and confirm there are no pending, " "in_progress, or failed items.\n" - "2. Verification — check the output against the original goal. For code tasks: " + "2. Verification: check the output against the original goal. For code tasks: " "run the tests or build with run_bash and confirm they pass. For research tasks: " "re-read the scratchpad and confirm the assembled answer addresses what was " "actually asked.\n" - "3. Uncertainty check — read the scratchpad and ask: are there unresolved " + "3. Uncertainty check: read the scratchpad and ask: are there unresolved " "questions, assumptions that were never validated, or tasks that were cancelled " "rather than properly completed?\n\n" "If all three are satisfied, give your final answer. If any are not, re-enter " - "the planning loop — add the outstanding items to the todo list and continue." + "the planning loop, add the outstanding items to the todo list and continue." ), } ] @@ -170,10 +170,10 @@ def agent_loop(client): messages.append(message) if message.tool_calls: - # The LLM wants to use one or more tools — run them, then loop + # The LLM wants to use one or more tools, run them, then loop handle_tool_calls(message.tool_calls, messages) elif not message.content or not message.content.strip(): - # The model ended its turn with an empty message — most commonly + # The model ended its turn with an empty message, most commonly # happens after a planning-only tool call (scratchpad / todo). # Nudge it to continue rather than silently stalling. messages.append({ diff --git a/agent-planning/tools/todo.py b/agent-planning/tools/todo.py index a11a0a9..4fcb6db 100644 --- a/agent-planning/tools/todo.py +++ b/agent-planning/tools/todo.py @@ -102,7 +102,7 @@ def todo_update(id, content=None, status=None) -> str: if item["status"] == "in_progress" and retries > 0: if retries >= RETRY_LIMIT: return ( - f"Updated to do item {id} to in_progress — " + f"Updated to do item {id} to in_progress, " f"but this is retry {retries} of { RETRY_LIMIT} (retry limit reached). " f"Do not retry again. Escalate to the user instead." diff --git a/agent-security/agent.py b/agent-security/agent.py index 36b94c3..a7b8ed3 100644 --- a/agent-security/agent.py +++ b/agent-security/agent.py @@ -95,7 +95,7 @@ def _ask_permission(tool_name: str, args: dict) -> bool: try: answer = input(" Allow this action? [y/n]: ").strip().lower() except EOFError: - print(" (EOF — denying permission)") + print(" (EOF, denying permission)") return False if answer in ("y", "yes"): return True @@ -118,9 +118,9 @@ def check_permission( Three layers, evaluated in order: - 1. Tool policy (§2.2 / §2.3) — a hard gate that no mode can + 1. Tool policy: a hard gate that no mode can override: path scoping, shell denylist, SSRF guard. - 2. Always-confirm patterns (§2.2) — destructive calls require + 2. Always-confirm patterns: destructive calls require explicit confirmation even in acceptEdits, and selected irreversible patterns refuse even in dangerouslySkipPermissions. 3. Mode-based decision (default / acceptEdits / skip). @@ -138,7 +138,7 @@ def check_permission( All other tools require explicit user approval. dangerouslySkipPermissions - All tools run without any prompt — EXCEPT calls blocked by layer + All tools run without any prompt, EXCEPT calls blocked by layer 1 (hard policy) or layer 2 (always-confirm / irreversible). """ # Layer 1: hard policy gate (path scope, shell, SSRF). @@ -171,11 +171,11 @@ def check_permission( if mode == PermissionMode.ACCEPT_EDITS and tool_name in WRITE_TOOLS: path = _resolve_tool_path(tool_name, args) if path and _is_within_working_dir(path, working_dir): - # an edit that empties an existing file is a delete — + # an edit that empties an existing file is a delete, # require confirmation rather than auto-approving. if _is_delete_via_write(tool_name, args, path): return _ask_permission(f"{tool_name} [DELETE-via-empty]", args), None - return True, None # auto-approved — within the working directory + return True, None # auto-approved, within the working directory # Path is outside the working directory → fall through to ask # Default mode, or acceptEdits for non-write / out-of-tree tools @@ -212,8 +212,8 @@ def build_tool_registry( Action tools (filesystem, shell, web) are dispatched into the long-lived Docker sandbox container. - If *allowed_tools* is given, only tools in that set are exposed — - this is the principle of least privilege (checklist §2.1). ``None`` + If *allowed_tools* is given, only tools in that set are exposed, + this is the principle of least privilege. ``None`` means "expose everything". """ registry = {} @@ -271,13 +271,13 @@ def handle_tool_calls( Every call passes through these layers, in order, and each decision is recorded in the audit log: - 0. Abort check (§6.3) + iteration cap (§4.1). - 1. Schema validation (§3.1/§3.3). - 2. Permission gate (§2.2/§2.3) — may prompt the user. + 0. Abort check + iteration cap. + 1. Schema validation. + 2. Permission gate, may prompt the user. 3. Execution: action tools run inside the Docker sandbox container; planning tools run in-process on the host. A container failure or timeout is reported back to the LLM as an error. - 4. File rollback snapshot (§6.3) before write/edit. + 4. File rollback snapshot before write/edit. """ caps = iteration_caps or IterationCaps() abort = abort_controller or AbortController() @@ -297,7 +297,7 @@ def handle_tool_calls( }) return - # §4.1 session-level tool-call cap. + # session-level tool-call cap. cap_reason = caps.bump_tool_call() if cap_reason is not None: audit.log("iteration_cap_hit", scope="session", @@ -349,7 +349,7 @@ def handle_tool_calls( # Schema-validate tool inputs: parse and validate before # any policy/permission check. A malformed call is reported # back to the LLM with the specific errors so it can correct - # and retry — it never reaches the sandbox or the permission + # and retry, it never reaches the sandbox or the permission # gate. try: args = json.loads(tool_call.function.arguments) @@ -460,7 +460,7 @@ def handle_tool_calls( name, args, mode, working_dir, ) - # §6.1 log the permission decision as a standalone event. + # log the permission decision as a standalone event. audit.log_permission_decision( tool=name, args=args, @@ -479,7 +479,7 @@ def handle_tool_calls( # File rollback: snapshot the original file before # any write/edit so the session can offer to revert on # abort. Only files inside the working dir are - # snapshotted (writes outside are already blocked by §2.3). + # snapshotted (writes outside are already blocked by). if name in ("write_file", "edit_file"): target_path = args.get("path", "") if target_path: @@ -495,7 +495,7 @@ def handle_tool_calls( "not execute this tool. Do not retry without changing " "the approach or asking the user." ) - # §4.3 distinguish timeouts from other container errors + # distinguish timeouts from other container errors # for the audit log. if "timed out" in str(e).lower(): audit.log("tool_timeout", tool=name, @@ -563,7 +563,7 @@ def agent_loop( container_name: str | None = None, approve_plan: bool = False, ): - # resource controls: the harness — never the model — owns these. + # resource controls: the harness, never the model, owns these. iteration_caps = iteration_caps or IterationCaps() context_budget = context_budget or ContextBudget() cost_tracker = cost_tracker or CostTracker() @@ -603,18 +603,18 @@ def agent_loop( "- Clarification (ask_question): ask the user a single focused question when you " "are genuinely blocked and cannot reasonably infer the missing information from " "context. Do not use it for progress updates or to confirm actions you can already " - "take — only ask when it is strictly necessary to proceed.\n\n" + "take, only ask when it is strictly necessary to proceed.\n\n" "## Execution environment (Docker sandbox)\n\n" - "Action tools — read_file, write_file, edit_file, glob_files, grep, run_bash, " - "webfetch — execute inside a Docker container. The user's project directory is " + "Action tools: read_file, write_file, edit_file, glob_files, grep, run_bash, " + "webfetch, execute inside a Docker container. The user's project directory is " "bind-mounted into that container at the same path as on the host, and the " "container's working directory is the project root. You can only see and modify " "files inside the project mount; the rest of the container's filesystem is a " "minimal, read-only-by-convention Linux image. If the session was started with " - "network disabled, run_bash and webfetch that need network will fail — treat that " + "network disabled, run_bash and webfetch that need network will fail, treat that " "as expected, not as a bug to work around. If a tool returns 'Container error: " - "...', do not retry the same call — adjust the approach or ask the user to run it " + "...', do not retry the same call, adjust the approach or ask the user to run it " "manually. Planning tools run on the host, not in the container, so their state " "persists across calls.\n\n" @@ -632,7 +632,7 @@ def agent_loop( "todo_append (status: pending).\n" "3. Before starting a step, mark it in_progress with todo_update. " "Keep only one item in_progress at a time.\n" - "4. Mark items done immediately after completing them — do not batch completions.\n" + "4. Mark items done immediately after completing them, do not batch completions.\n" "5. Call todo_list to review remaining work before moving to the next step.\n" "6. Mark tasks cancelled if they become unnecessary.\n\n" @@ -640,16 +640,16 @@ def agent_loop( "Planning tool calls (write_scratchpad, todo_append, todo_update, todo_list) " "are internal bookkeeping, not responses to the user. After any planning tool " - "call, always continue working immediately — make your next tool call or, once " + "call, always continue working immediately, make your next tool call or, once " "the task is fully complete, give a substantive final answer. " "Never emit an empty or whitespace-only message.\n\n" "## Replanning\n\n" "After every tool result, check whether the outcome matched your expectation. " "If a tool returns an error, unexpected output, or reveals information that " - "changes your understanding of the task, do not move to the next planned step — " + "changes your understanding of the task, do not move to the next planned step, " "replan first.\n\n" "When a step fails:\n" - "1. Diagnose in the scratchpad — is this a recoverable input error (wrong path, " + "1. Diagnose in the scratchpad: is this a recoverable input error (wrong path, " "typo, wrong argument) or a deeper problem (wrong approach, wrong assumption)?\n" "2. Mark the task failed: todo_update(id, status='failed').\n" "3. Choose a recovery action:\n" @@ -659,8 +659,8 @@ def agent_loop( " - Reorder: new information makes a different task more urgent. Update the " "pending items before continuing.\n" "4. If todo_update reports that the retry limit has been reached, stop retrying. " - "Write a clear diagnosis in the scratchpad — what you tried, what failed each " - "time, and what you need — then give the user a concise escalation message " + "Write a clear diagnosis in the scratchpad: what you tried, what failed each " + "time, and what you need, then give the user a concise escalation message " "and wait for their input.\n\n" "When a tool succeeds but returns information that changes the picture, pause " "before acting. Call todo_list, reassess all pending items in the scratchpad, " @@ -668,34 +668,34 @@ def agent_loop( "## How to use the scratchpad\n\n" "Before each tool call during a complex task, update the scratchpad with your " "current thinking. Structure each entry around these five steps:\n\n" - "1. Restate the goal — write what you understand the task to be, in your own words. " + "1. Restate the goal: write what you understand the task to be, in your own words. " "This catches misreads before they compound into wasted work.\n" - "2. Survey what you know — note which files you have seen, what the code structure " + "2. Survey what you know: note which files you have seen, what the code structure " "looks like, and what constraints or requirements apply.\n" - "3. Evaluate options — reason through at least two approaches and explain why you " + "3. Evaluate options: reason through at least two approaches and explain why you " "are choosing one over the other (e.g. 'I could rewrite the middleware, or wrap it. " "Wrapping is safer because it leaves the existing call sites untouched.').\n" - "4. Anticipate failure modes — write down what could go wrong with the chosen " + "4. Anticipate failure modes: write down what could go wrong with the chosen " "approach and how you would diagnose it (e.g. 'If the tests fail after this, the " "most likely cause is that the session cookie name changed.').\n" - "5. Decide the next single action — commit to exactly one tool call. " + "5. Decide the next single action: commit to exactly one tool call. " "Do not plan several calls at once; decide the next step only.\n\n" "Re-read the scratchpad whenever you resume after a tool result to keep your " "reasoning grounded in what you have already learned.\n\n" "## Done detection\n\n" "Do not give a final answer based on the task list being empty alone. " "Before declaring the task complete, verify all three of the following:\n\n" - "1. Structural completion — call todo_list and confirm there are no pending, " + "1. Structural completion: call todo_list and confirm there are no pending, " "in_progress, or failed items.\n" - "2. Verification — check the output against the original goal. For code tasks: " + "2. Verification: check the output against the original goal. For code tasks: " "run the tests or build with run_bash and confirm they pass. For research tasks: " "re-read the scratchpad and confirm the assembled answer addresses what was " "actually asked.\n" - "3. Uncertainty check — read the scratchpad and ask: are there unresolved " + "3. Uncertainty check: read the scratchpad and ask: are there unresolved " "questions, assumptions that were never validated, or tasks that were cancelled " "rather than properly completed?\n\n" "If all three are satisfied, give your final answer. If any are not, re-enter " - "the planning loop — add the outstanding items to the todo list and continue." + "the planning loop, add the outstanding items to the todo list and continue." ) _prompt_warnings = secret_management.audit_system_prompt( _system_prompt_body) @@ -795,7 +795,7 @@ def agent_loop( messages=messages, tools=active_schemas, temperature=0.7, - timeout=llm_timeout, # §4.3 LLM call timeout + timeout=llm_timeout, ) except Exception as e: # surface an LLM-call failure (including timeout) @@ -807,7 +807,7 @@ def agent_loop( break # record usage for cost tracking. Local backends - # (Ollama) may not populate usage; that's fine — the + # (Ollama) may not populate usage; that's fine, the # tracker simply records zero tokens for the call. cost_tracker.record_usage(getattr(response, "usage", None)) @@ -845,7 +845,7 @@ def agent_loop( audit.log_assistant_message(message.content, message.tool_calls) if message.tool_calls: - # The LLM wants to use one or more tools — run them, then loop + # The LLM wants to use one or more tools, run them, then loop handle_tool_calls( message.tool_calls, messages, @@ -862,7 +862,7 @@ def agent_loop( plan_state=plan_state, ) elif not message.content or not message.content.strip(): - # The model ended its turn with an empty message — most commonly + # The model ended its turn with an empty message, most commonly # happens after a planning-only tool call (scratchpad / todo). # Nudge it to continue rather than silently stalling. messages.append({ @@ -876,7 +876,7 @@ def agent_loop( # needed here. If a web UI is added in the future, the # assistant content MUST be passed through html.escape # (or a template engine's auto-escaping) before being - # inserted into the DOM — never trust model output. + # inserted into the DOM, never trust model output. print(f"Assistant: {message.content}") break @@ -930,7 +930,7 @@ if __name__ == "__main__": default=None, help=( "Comma-separated allowlist of tool names to expose to the agent " - "(principle of least privilege, checklist §2.1). " + "(principle of least privilege). " "Example: --tools read_file,glob_files,grep,run_bash. " "Default: all tools are exposed." ), @@ -940,7 +940,7 @@ if __name__ == "__main__": type=float, default=120.0, help=( - "§4.3 Per-tool-call timeout in seconds. A tool that does not " + "Per-tool-call timeout in seconds. A tool that does not " "finish in this time is killed and reported to the LLM as a " "timeout. Default: 120." ), @@ -950,7 +950,7 @@ if __name__ == "__main__": type=float, default=120.0, help=( - "§4.3 Timeout for each LLM completion call in seconds. " + "Timeout for each LLM completion call in seconds. " "Default: 120." ), ) @@ -959,7 +959,7 @@ if __name__ == "__main__": type=int, default=None, help=( - "§4.1 Max LLM turns per user message before the loop stops " + "Max LLM turns per user message before the loop stops " f"and asks for a summary. Default: { resource_limits.DEFAULT_MAX_TURNS_PER_USER_MSG}." ), @@ -969,7 +969,7 @@ if __name__ == "__main__": type=int, default=None, help=( - "§4.1 Max tool calls for the whole session. Default: " + "Max tool calls for the whole session. Default: " f"{resource_limits.DEFAULT_MAX_TOOL_CALLS_PER_SESSION}." ), ) @@ -978,7 +978,7 @@ if __name__ == "__main__": type=int, default=None, help=( - "§4.2 Token budget for the message history. When exceeded, " + "Token budget for the message history. When exceeded, " "older messages are trimmed and replaced with a summary. " f"Default: {resource_limits.DEFAULT_MAX_CONTEXT_TOKENS}." ), @@ -988,7 +988,7 @@ if __name__ == "__main__": type=float, default=None, help=( - "§4.4 Cumulative API spend cap in USD. When exceeded, the " + "Cumulative API spend cap in USD. When exceeded, the " "session aborts gracefully. Default: " f"{resource_limits.DEFAULT_MAX_COST_USD:.2f} (no effect for " "local backends that don't report usage)." @@ -999,7 +999,7 @@ if __name__ == "__main__": action="store_true", default=False, help=( - "§6.2 Require human approval of the agent's plan before any " + "Require human approval of the agent's plan before any " "action tool runs. The agent builds its plan in the " "scratchpad + todo list; the first time it tries to run an " "action tool, the user is shown the plan and asked to approve." @@ -1038,7 +1038,7 @@ if __name__ == "__main__": print(" These will NOT be passed to the sandbox container " "(only an allowlist is inherited).") - # 5.2 Verify no credential mount paths would leak into the container. + # Verify no credential mount paths would leak into the container. cred_mounts = secret_management.check_credential_mounts() if cred_mounts: print( @@ -1057,8 +1057,8 @@ if __name__ == "__main__": project_root=sandbox_root, tools_dir=tools_dir, network=cli_args.network, - exec_timeout=cli_args.tool_timeout, # §4.3 - container_env=container_env, # §5.2 / §5.3 + exec_timeout=cli_args.tool_timeout, + container_env=container_env, ) except DockerSandboxError as e: print(f"Could not start sandbox: {e}") @@ -1096,10 +1096,10 @@ if __name__ == "__main__": max_tool_calls_per_session=iteration_caps.max_tool_calls_per_session, max_context_tokens=context_budget.max_tokens, max_cost_usd=cost_tracker.max_cost_usd, - session_id=session_creds.session_id, # §5.3 - secret_env_count=len(secret_env), # §5.1 - credential_mounts_found=[str(p) for p in cred_mounts], # §5.2 - container_env_allowlist=sorted(container_env.keys()), # §5.2 + session_id=session_creds.session_id, + secret_env_count=len(secret_env), + credential_mounts_found=[str(p) for p in cred_mounts], + container_env_allowlist=sorted(container_env.keys()), ) print(f"Agent started in '{ @@ -1117,10 +1117,10 @@ if __name__ == "__main__": print(f"Audit log: {audit.path}") print("Type \\exit to quit. (Ctrl-C to abort a session)\n") - # session abort controller — installs SIGINT/SIGTERM handlers. + # session abort controller: installs SIGINT/SIGTERM handlers. abort_controller = AbortController() abort_controller.install_signal_handlers() - # file rollback — snapshots before write/edit for revert on abort. + # file rollback: snapshots before write/edit for revert on abort. file_rollback = FileRollback() client = get_llm_client() @@ -1144,7 +1144,7 @@ if __name__ == "__main__": if file_rollback.snapshot_count > 0: file_rollback.offer_rollback() file_rollback.cleanup() - session_creds.revoke() # §5.3 rotate on exit + session_creds.revoke() sandbox.close() audit.close() print(f"\nSandbox container removed. Audit log written to { diff --git a/agent-security/prompt_safety.py b/agent-security/prompt_safety.py index 80a1f60..a3bc749 100644 --- a/agent-security/prompt_safety.py +++ b/agent-security/prompt_safety.py @@ -110,7 +110,7 @@ def mark_external_content( their content in an ```` tag so the model treats them as data rather than instructions. - Error strings from the tools are returned unchanged — they are + Error strings from the tools are returned unchanged, they are harness-generated, not external content. """ if tool_name == "webfetch": diff --git a/agent-security/resource_limits.py b/agent-security/resource_limits.py index de40adb..d3d78d3 100644 --- a/agent-security/resource_limits.py +++ b/agent-security/resource_limits.py @@ -1,14 +1,14 @@ -"""Loop & resource controls (checklist §4). +"""Loop & resource controls. This module centralises every limit that prevents the agent from running away with itself: - §4.1 IterationCaps — hard caps on turns and tool calls. - §4.2 ContextBudget — token budget with automatic context trimming. - §4.3 (timeouts live in sandbox.py / agent.py — wired here via a small + IterationCaps: hard caps on turns and tool calls. + ContextBudget: token budget with automatic context trimming. + (timeouts live in sandbox.py / agent.py, wired here via a small ToolTimeout helper that classifies timeout events for the audit log). - §4.4 CostTracker — cumulative API spend circuit breaker. + CostTracker: cumulative API spend circuit breaker. All of these are *host-side* controls: the model never gets to vote on them. They are evaluated between tool calls and before each LLM call. @@ -22,7 +22,7 @@ from typing import Any # --------------------------------------------------------------------------- -# 4.1 Hard iteration caps +# Hard iteration caps # --------------------------------------------------------------------------- # Defaults are chosen to be generous enough for real coding tasks but @@ -33,9 +33,9 @@ DEFAULT_MAX_TOOL_CALLS_PER_SESSION = 200 @dataclass class IterationCaps: - """Counters that enforce hard iteration caps (§4.1). + """Counters that enforce hard iteration caps. - The harness — never the model — owns these limits. Two counters + The harness, never the model, owns these limits. Two counters are tracked: - ``turns``: incremented once per LLM response within a single @@ -82,11 +82,11 @@ class IterationCaps: @property def breached(self) -> bool: return self.turns > self.max_turns_per_user_msg or \ - self.tool_calls > self.max_tool_calls_per_session + self.tool_calls > self.max_tool_calls_per_session # --------------------------------------------------------------------------- -# 4.2 Token budget enforcement +# Token budget enforcement # --------------------------------------------------------------------------- # Above this fraction of the model's context window we trim older @@ -102,7 +102,7 @@ def estimate_tokens(text: str) -> int: """Cheap token estimate: ~4 chars per token. Good enough for budget decisions; the real BPE count is only - needed for billing, which §4.4 handles via the API's own + needed for billing, which handles via the API's own ``usage`` field. """ if not text: @@ -112,7 +112,8 @@ def estimate_tokens(text: str) -> int: def _message_token_count(msg: Any) -> int: """Estimate tokens in a single chat message.""" - content = msg.get("content") if isinstance(msg, dict) else getattr(msg, "content", None) + content = msg.get("content") if isinstance( + msg, dict) else getattr(msg, "content", None) if content is None: return 0 if isinstance(content, str): @@ -124,7 +125,7 @@ def _message_token_count(msg: Any) -> int: @dataclass class ContextBudget: - """Track cumulative tokens and trim the message history (§4.2). + """Track cumulative tokens and trim the message history. ``check_and_trim`` is called before each LLM request. If the estimated token count exceeds ``max_tokens * trim_threshold`` it @@ -187,10 +188,12 @@ class ContextBudget: tool_calls: list[str] = [] total_chars = 0 for m in dropped: - content = m.get("content") if isinstance(m, dict) else getattr(m, "content", "") + content = m.get("content") if isinstance( + m, dict) else getattr(m, "content", "") if content: total_chars += len(str(content)) - tcs = m.get("tool_calls") if isinstance(m, dict) else getattr(m, "tool_calls", None) + tcs = m.get("tool_calls") if isinstance( + m, dict) else getattr(m, "tool_calls", None) if tcs: for tc in tcs: name = getattr(getattr(tc, "function", None), "name", None) @@ -201,12 +204,14 @@ class ContextBudget: blob = str([ (m.get("role") if isinstance(m, dict) else getattr(m, "role", "")) for m in dropped ]) - digest = hashlib.sha256(blob.encode("utf-8", "replace")).hexdigest()[:16] + digest = hashlib.sha256(blob.encode( + "utf-8", "replace")).hexdigest()[:16] lines = [ "[context-trim summary]", f"Earlier conversation dropped to stay within the token budget.", f"Dropped messages: {len(dropped)} ({total_chars} chars).", - f"Tools called in dropped section: {', '.join(tool_calls) or 'none'}.", + f"Tools called in dropped section: { + ', '.join(tool_calls) or 'none'}.", f"Conversation hash (first 16 hex): {digest}.", "Re-read any files you need rather than relying on the dropped context.", ] @@ -214,7 +219,7 @@ class ContextBudget: def cap_tool_result(result: str, limit: int = DEFAULT_MAX_TOOL_RESULT_CHARS) -> str: - """Cap a tool result before it is inserted into messages (§4.2). + """Cap a tool result before it is inserted into messages. Long results (e.g. reading a 50k-line file) are truncated to ``limit`` chars with a notice appended so the model knows there is @@ -230,7 +235,7 @@ def cap_tool_result(result: str, limit: int = DEFAULT_MAX_TOOL_RESULT_CHARS) -> # --------------------------------------------------------------------------- -# 4.4 Cost circuit breakers +# Cost circuit breakers # --------------------------------------------------------------------------- # Default per-session spend cap in USD. Generous for local Ollama @@ -246,7 +251,7 @@ DEFAULT_PRICE_PER_1K_OUT = 0.000600 @dataclass class CostTracker: - """Accumulate API spend and abort when the cap is hit (§4.4). + """Accumulate API spend and abort when the cap is hit. After each ``chat.completions.create`` call, the caller invokes ``record_usage`` with the ``response.usage`` object (or None for @@ -292,7 +297,8 @@ class CostTracker: def summary(self) -> str: return ( f"calls={self.calls} tokens_in={self.total_tokens_in} " - f"tokens_out={self.total_tokens_out} cost=${self.total_cost_usd:.4f}" + f"tokens_out={self.total_tokens_out} cost=${ + self.total_cost_usd:.4f}" ) diff --git a/agent-security/secret_management.py b/agent-security/secret_management.py index ab1d5cb..f24880c 100644 --- a/agent-security/secret_management.py +++ b/agent-security/secret_management.py @@ -1,22 +1,22 @@ -"""Secret & credential management (checklist §5). +"""Secret & credential management. Three layers: - §5.1 Never put secrets in the system prompt. + Never put secrets in the system prompt. - ``scan_environment_for_secrets`` runs at startup and warns about env vars whose names look like credentials. - ``audit_system_prompt`` statically checks a prompt string for f-string interpolation of env vars or known secret names. - §5.2 Credential injection at the harness level. + Credential injection at the harness level. - ``build_container_env`` returns the minimal environment dict - passed to the sandbox container — only an allowlist of vars + passed to the sandbox container: only an allowlist of vars the agent genuinely needs, never raw host credentials. - ``CREDENTIAL_MOUNT_PATHS`` lists host files/dirs that must NOT be bind-mounted into the container (``~/.aws``, ``~/.ssh``, ``~/.netrc``, …). - §5.3 Rotate credentials per session. + Rotate credentials per session. - ``SessionCredentials`` generates a fresh per-session token via ``secrets.token_urlsafe``; the harness can pass it to the container and use it to authenticate any harness-side @@ -34,12 +34,12 @@ from typing import Any # --------------------------------------------------------------------------- -# 5.1 Never put secrets in the system prompt +# Never put secrets in the system prompt # --------------------------------------------------------------------------- # Env-var name pattern that *looks* like a secret. Matching is # case-insensitive. False positives (e.g. ``PRINTER_SETTINGS``) are -# fine — we only warn, we don't block. +# fine, we only warn, we don't block. SECRET_ENV_PATTERN = re.compile( r"(.*(?:KEY|SECRET|TOKEN|PASSWORD|PASSWD|CREDENTIAL|APIKEY|API_KEY|AUTH).*)", re.IGNORECASE, @@ -58,7 +58,7 @@ def scan_environment_for_secrets() -> list[tuple[str, str]]: """Return a list of ``(name, source)`` for env vars that look like secrets. ``source`` is ``"env"`` (a real environment variable). This is a - startup warning only — it does not modify the environment. The + startup warning only, it does not modify the environment. The caller (the harness) decides whether to scrub them before starting the sandbox container (see ``build_container_env``). """ @@ -74,9 +74,9 @@ def audit_system_prompt(prompt: str) -> list[str]: Returns a list of warnings (empty if clean). This catches: - - ``f"... {os.environ['API_KEY']} ..."`` — interpolating env + - ``f"... {os.environ['API_KEY']} ..."``: interpolating env vars directly into the prompt. - - ``f"... {os.getenv('SECRET')} ..."`` — same, via getenv. + - ``f"... {os.getenv('SECRET')} ..."``: same, via getenv. - Literal occurrences of known secret-looking env-var names. """ warnings: list[str] = [] @@ -84,7 +84,7 @@ def audit_system_prompt(prompt: str) -> list[str]: if PROMPT_INTERPOLATION_PATTERN.search(prompt): warnings.append( "System prompt appears to interpolate os.environ / os.getenv. " - "Never put secrets in the system prompt — the model can leak " + "Never put secrets in the system prompt, the model can leak " "them in tool calls or responses." ) @@ -101,7 +101,7 @@ def audit_system_prompt(prompt: str) -> list[str]: # --------------------------------------------------------------------------- -# 5.2 Credential injection at the harness level +# Credential injection at the harness level # --------------------------------------------------------------------------- # Host paths that must NEVER be bind-mounted into the sandbox container. @@ -130,8 +130,8 @@ ALLOWED_CONTAINER_ENV = frozenset({ "LANG", "LC_ALL", "TERM", - "AGENT_SESSION_ID", # set per-session by the harness (§5.3) - "AGENT_SESSION_TOKEN", # short-lived, per-session (§5.3) + "AGENT_SESSION_ID", # set per-session by the harness + "AGENT_SESSION_TOKEN", # short-lived, per-session }) @@ -162,7 +162,7 @@ def build_container_env(session_id: str, session_token: str) -> dict[str, str]: # --------------------------------------------------------------------------- -# 5.3 Rotate credentials per session +# Rotate credentials per session # --------------------------------------------------------------------------- class SessionCredentials: diff --git a/agent-security/session_control.py b/agent-security/session_control.py index 4acbf3c..6c5380a 100644 --- a/agent-security/session_control.py +++ b/agent-security/session_control.py @@ -1,15 +1,15 @@ -"""Session-level abort & kill switches (checklist §6.3). +"""Session-level abort & kill switches. Three pieces: - 1. ``AbortController`` — a thread-safe flag set by a signal handler + 1. ``AbortController``: a thread-safe flag set by a signal handler (SIGINT / SIGTERM) or by any code path that detects a fatal condition (cost cap, iteration cap, user request). - 2. ``FileRollback`` — snapshots original file bytes before each + 2. ``FileRollback``: snapshots original file bytes before each write/edit so that an abort can offer to revert reversible state. - 3. ``kill_in_flight`` — a helper that sends SIGINT to any python + 3. ``kill_in_flight``: a helper that sends SIGINT to any python process running inside the sandbox container, stopping a hanging ``docker exec`` without tearing down the container itself. @@ -81,7 +81,7 @@ class AbortController: """Register SIGINT / SIGTERM handlers that call ``trigger``. Only call this from the main thread (signal.signal requirement). - The previous handlers are not saved — we deliberately replace + The previous handlers are not saved, we deliberately replace them because the abort controller is the final arbiter. """ def _handler(signum, frame): @@ -94,7 +94,7 @@ class AbortController: self._registered_signals.append(sig) except (ValueError, OSError): # Not in main thread, or signal not supported on this - # platform — skip silently. + # platform, skip silently. pass def remove_signal_handlers(self) -> None: @@ -126,7 +126,7 @@ def kill_in_flight(container: str) -> None: ) except Exception: # Best-effort: if the container is already gone or pkill isn't - # available, the exec subprocess's own timeout (§4.3) will + # available, the exec subprocess's own timeout will # eventually clean up. pass @@ -136,14 +136,14 @@ def kill_in_flight(container: str) -> None: # --------------------------------------------------------------------------- class FileRollback: - """Snapshot original file bytes before each write/edit (§6.3). + """Snapshot original file bytes before each write/edit. On abort, ``offer_rollback`` walks the snapshot list and restores each file to its pre-edit state, prompting the user for confirmation. Only ``write_file`` and ``edit_file`` are reversible; ``run_bash`` - side effects (e.g. ``git commit``) are not — the audit log is the + side effects (e.g. ``git commit``) are not, the audit log is the only record for those. """ @@ -154,7 +154,8 @@ class FileRollback: def _ensure_backup_dir(self) -> Path: if self._backup_dir is None: - self._backup_dir = Path(__file__).resolve().parent / ".rollback_backups" + self._backup_dir = Path( + __file__).resolve().parent / ".rollback_backups" self._backup_dir.mkdir(parents=True, exist_ok=True) return self._backup_dir @@ -171,7 +172,7 @@ class FileRollback: shutil.copy2(p, backup) self._snapshots.append((str(p.resolve()), backup)) except OSError: - # If we can't snapshot, we just can't roll back — don't + # If we can't snapshot, we just can't roll back, don't # block the tool call. pass diff --git a/agent-security/tool_policy.py b/agent-security/tool_policy.py index 54d8320..d2b7207 100644 --- a/agent-security/tool_policy.py +++ b/agent-security/tool_policy.py @@ -1,7 +1,7 @@ """Tool permission & parameter-scoping policy. -Implements checklist §2.2 (confirmation for destructive actions) and -§2.3 (scope tool parameters) in one place so the rules are easy to +Implements confirmation for destructive actions and +scope tool parameters in one place so the rules are easy to audit and extend. Three layers, evaluated in order by ``check_tool_policy`` before the @@ -42,14 +42,14 @@ DESTRUCTIVE_TOOLS = frozenset({ # Tools that ALWAYS require explicit human confirmation, even under # ``dangerouslySkipPermissions``. These are the irreversible / exfil -# class — we refuse them outright (a denylist), not just prompt for them. +# class, we refuse them outright (a denylist), not just prompt for them. # # ``run_bash`` is not in this set as a whole; instead its *command* is # screened by the shell policy below. ``webfetch`` is also policy- # screened (SSRF). This set is reserved for tools where *any* call is # too dangerous to auto-run. ALWAYS_CONFIRM_TOOLS = frozenset({ - # Currently empty — kept for future destructive tools (e.g. delete_file, + # Currently empty, kept for future destructive tools (e.g. delete_file, # send_email). The shell policy handles the dangerous run_bash cases. }) @@ -160,7 +160,7 @@ SHELL_DENYLIST_BINARIES = frozenset({ "docker", # sandbox escape / host control "sudo", "su", # privilege escalation "nc", "netcat", "ncat", # reverse shells / exfil - "curl", "wget", # exfil / SSRF — handled here AND in web policy + "curl", "wget", # exfil / SSRF: handled here AND in web policy "chmod", "chown", # permission tampering "mkfs", "dd", # destructive disk ops "shutdown", "reboot", "halt", "poweroff", @@ -200,7 +200,7 @@ def check_shell_policy(command: str) -> tuple[bool, str | None]: try: tokens = shlex.split(command) except ValueError: - # Unparseable (e.g. unbalanced quotes) — let the shell itself + # Unparseable (e.g. unbalanced quotes), let the shell itself # reject it, but flag for confirmation. return False, "Shell command could not be parsed (unbalanced quotes)." @@ -266,7 +266,7 @@ def check_web_policy(url: str) -> tuple[bool, str | None]: if addr.is_loopback or addr.is_link_local or addr.is_multicast: return False, f"Blocked IP '{ip}' (loopback / link-local / multicast)." if addr.is_private: - return False, f"Blocked IP '{ip}' (RFC1918 private range — SSRF guard)." + return False, f"Blocked IP '{ip}' (RFC1918 private range: SSRF guard)." return True, None diff --git a/agent-security/tools/_dispatch.py b/agent-security/tools/_dispatch.py index 212839a..d2cb824 100644 --- a/agent-security/tools/_dispatch.py +++ b/agent-security/tools/_dispatch.py @@ -6,7 +6,7 @@ Invoked by the host as: The tool arguments are read from stdin as a JSON object; the result is printed to stdout. Only the *action* tools (the ones that touch the -filesystem, shell, or network) are dispatched here — the in-memory +filesystem, shell, or network) are dispatched here: the in-memory planning tools stay on the host. """ diff --git a/agent-security/tools/audit.py b/agent-security/tools/audit.py index beed1c8..22e8552 100644 --- a/agent-security/tools/audit.py +++ b/agent-security/tools/audit.py @@ -5,13 +5,13 @@ ISO timestamp and session id. The log is flushed after every record so that a crash still leaves a complete trail. The file alone is enough to reconstruct what the agent did, in order. -§2.4: tool results are truncated to ``MAX_RESULT_BYTES`` in the log to +tool results are truncated to ``MAX_RESULT_BYTES`` in the log to prevent unbounded growth; a SHA-256 and full length are stored alongside so the truncated entry is self-describing and tamper-evident. -§6.1: dedicated events for every decision step — +dedicated events for every decision step, ``log_permission_decision``, ``log_llm_request``, ``log_llm_response``, -``log_session_abort`` — give full forensic replay without guessing. +``log_session_abort``, give full forensic replay without guessing. """ import hashlib @@ -66,7 +66,8 @@ class AuditLog: "event": event, } record.update(fields) - self._fh.write(json.dumps(record, ensure_ascii=False, default=str) + "\n") + self._fh.write(json.dumps( + record, ensure_ascii=False, default=str) + "\n") self._fh.flush() # -- convenience wrappers ------------------------------------------ @@ -86,7 +87,8 @@ class AuditLog: args = json.loads(tc.function.arguments) except Exception: args = tc.function.arguments - calls.append({"id": tc.id, "name": tc.function.name, "args": args}) + calls.append( + {"id": tc.id, "name": tc.function.name, "args": args}) self.log("assistant_message", content=content, tool_calls=calls) def log_tool_result( @@ -116,8 +118,6 @@ class AuditLog: **_truncate_for_log(result), ) - # -- §6.1 decision-step logging ---------------------------------- - def log_permission_decision( self, tool: str, @@ -126,7 +126,7 @@ class AuditLog: allowed: bool, reason: str | None = None, ) -> None: - """Record a standalone permission decision (§6.1). + """Record a standalone permission decision. This is emitted *before* the tool runs (or is refused), so the audit trail shows the decision and its reason even if the @@ -148,7 +148,7 @@ class AuditLog: token_estimate: int, has_tools: bool, ) -> None: - """Record that an LLM request is about to be sent (§6.1).""" + """Record that an LLM request is about to be sent.""" self.log( "llm_request", model=model, @@ -165,7 +165,7 @@ class AuditLog: message_hash: str | None = None, tool_call_count: int = 0, ) -> None: - """Record the LLM response metadata (§6.1). + """Record the LLM response metadata. ``message_hash`` is a SHA-256 of the assistant message content so the full conversation can be verified for forensic replay @@ -181,7 +181,7 @@ class AuditLog: ) def log_session_abort(self, reason: str) -> None: - """Record that the session was aborted (§6.3).""" + """Record that the session was aborted.""" self.log("session_abort", reason=reason) def close(self) -> None: diff --git a/agent-security/tools/interaction.py b/agent-security/tools/interaction.py index 34a3540..6af941e 100644 --- a/agent-security/tools/interaction.py +++ b/agent-security/tools/interaction.py @@ -4,5 +4,5 @@ def ask_question(question: str) -> str: try: answer = input(" Your answer: ").strip() except EOFError: - return "(no answer — EOF)" + return "(no answer, EOF)" return answer if answer else "(no answer provided)" diff --git a/agent-security/tools/registry.py b/agent-security/tools/registry.py index dbf7c40..64197c0 100644 --- a/agent-security/tools/registry.py +++ b/agent-security/tools/registry.py @@ -265,7 +265,7 @@ def get_tool_schemas(): "and cannot reasonably infer it from context. " "Ask one focused question at a time. " "Do not use this for progress updates or to confirm actions you can already " - "take — only ask when you are genuinely blocked." + "take, only ask when you are genuinely blocked." ), "parameters": { "type": "object", diff --git a/agent-security/tools/sandbox.py b/agent-security/tools/sandbox.py index 5541541..f832749 100644 --- a/agent-security/tools/sandbox.py +++ b/agent-security/tools/sandbox.py @@ -11,12 +11,12 @@ read-only to the tool. Network egress can be disabled entirely with The container is started once per session and reused for every action tool call (via ``docker exec``) to avoid per-call startup latency. In-memory planning tools (todo, scratchpad, ask_question) are **not** -run in the container — their state would not survive between separate -``docker exec`` processes — so they stay in-process on the host. +run in the container, their state would not survive between separate +``docker exec`` processes, so they stay in-process on the host. This module requires a Docker-compatible container CLI on the host: Docker, or Podman (which is auto-detected when ``docker info`` does not -work — note that a shell ``alias docker=podman`` is **not** enough, +work, note that a shell ``alias docker=podman`` is **not** enough, because the agent invokes the binary directly via ``subprocess`` without a shell). The runtime can also be forced with the ``$AGENT_DOCKER`` environment variable. On first use the ``agent-security-runner`` image @@ -132,7 +132,7 @@ class DockerSandbox: self.image = image self.network = network self.exec_timeout = float(exec_timeout) - # the harness — not the model — controls the container env. + # the harness, not the model, controls the container env. # Only an allowlist of vars is inherited from the host; secret- # looking env vars are stripped before the container ever starts. self.container_env = container_env or {} @@ -221,7 +221,7 @@ class DockerSandbox: def run_tool(self, name: str, args: dict) -> str: """Execute *name* with *args* inside the container, return its output. - §4.3: enforces a per-call timeout (``self.exec_timeout``). A + enforces a per-call timeout (``self.exec_timeout``). A timeout is reported back as a ``DockerSandboxError`` with a clear message so the LLM knows not to retry blindly. """ @@ -241,7 +241,7 @@ class DockerSandbox: raise DockerSandboxError( f"Tool '{name}' timed out after {self.exec_timeout:.0f}s. " "The command did not finish in the allowed time. Do not " - "retry the same call — adjust the approach or ask the user." + "retry the same call, adjust the approach or ask the user." ) if proc.returncode != 0: err = (proc.stderr or proc.stdout or "").strip() diff --git a/agent-security/tools/todo.py b/agent-security/tools/todo.py index a11a0a9..4fcb6db 100644 --- a/agent-security/tools/todo.py +++ b/agent-security/tools/todo.py @@ -102,7 +102,7 @@ def todo_update(id, content=None, status=None) -> str: if item["status"] == "in_progress" and retries > 0: if retries >= RETRY_LIMIT: return ( - f"Updated to do item {id} to in_progress — " + f"Updated to do item {id} to in_progress, " f"but this is retry {retries} of { RETRY_LIMIT} (retry limit reached). " f"Do not retry again. Escalate to the user instead." diff --git a/agent-security/tools/validators.py b/agent-security/tools/validators.py index ebeac30..6d7cb66 100644 --- a/agent-security/tools/validators.py +++ b/agent-security/tools/validators.py @@ -1,4 +1,4 @@ -"""Lightweight JSON-Schema validator for tool inputs (checklist §3.1, §3.3). +"""Lightweight JSON-Schema validator for tool inputs. We deliberately avoid a third-party dependency (``jsonschema`` / ``pydantic``) so the host-side validation needs no new install and no @@ -12,7 +12,7 @@ JSON-Schema Draft 7 actually used by ``tools/registry.get_tool_schemas``: - ``minimum`` / ``maximum`` - ``minLength`` / ``maxLength`` -§3.3 (limit output scope) is enforced by the same schemas: bounds on +limit output scope is enforced by the same schemas: bounds on ``offset``, ``limit``, ``command`` length, ``content`` length, and a rejection of absolute-path glob patterns are baked into the schemas and therefore checked here. @@ -43,7 +43,7 @@ def _check_type(value: Any, expected: str) -> str | None: if not isinstance(value, str): return f"expected string, got {type(value).__name__}" elif expected == "integer": - # bool is a subclass of int — reject it explicitly. + # bool is a subclass of int, reject it explicitly. if isinstance(value, bool) or not isinstance(value, int): return f"expected integer, got {type(value).__name__}" elif expected == "boolean": @@ -58,7 +58,7 @@ def validate_args(args: dict[str, Any], schema: dict) -> tuple[bool, list[str]]: """Validate *args* against a tool's JSON-Schema function spec. ``schema`` is the inner ``{"type": "object", "properties": ...}`` - dict — i.e. ``tool["function"]["parameters"]``. + dict, i.e. ``tool["function"]["parameters"]``. Returns ``(ok, errors)``. """ @@ -89,11 +89,11 @@ def validate_args(args: dict[str, Any], schema: dict) -> tuple[bool, list[str]]: # --------------------------------------------------------------------------- -# Bounded schemas (§3.3 limit output scope) +# Bounded schemas # --------------------------------------------------------------------------- def bounded_schemas(raw_schemas: list[dict]) -> list[dict]: - """Return a copy of *raw_schemas* with §3.3 bounds injected. + """Return a copy of *raw_schemas* with bounds injected. We mutate copies of the per-tool parameter schemas to add: @@ -104,7 +104,7 @@ def bounded_schemas(raw_schemas: list[dict]) -> list[dict]: - ``run_bash.command``: minLength 1, maxLength 4096 - ``webfetch.url``: maxLength 4096 - ``glob_files.pattern``: reject absolute paths (pattern check - implemented in ``_validate_value`` via a custom constraint — + implemented in ``_validate_value`` via a custom constraint, here we add ``format: relative-path`` which our validator treats specially). @@ -142,7 +142,8 @@ def bounded_schemas(raw_schemas: list[dict]) -> list[dict]: # would escape the working-dir scoping. props.setdefault("pattern", {})["format"] = "relative-path" - out.append({"type": "function", "function": {**fn, "parameters": params}}) + out.append({"type": "function", "function": { + **fn, "parameters": params}}) return out @@ -181,12 +182,15 @@ def _validate_value(value: Any, schema: dict, path: str) -> list[str]: if schema.get("type") == "string": if "minLength" in schema and len(value) < schema["minLength"]: - errs.append(f"{path}: length {len(value)} < minLength {schema['minLength']}") + errs.append(f"{path}: length {len(value)} < minLength { + schema['minLength']}") if "maxLength" in schema and len(value) > schema["maxLength"]: - errs.append(f"{path}: length {len(value)} > maxLength {schema['maxLength']}") - # §3.3 custom format: relative-path (reject absolute glob patterns). + errs.append(f"{path}: length {len(value)} > maxLength { + schema['maxLength']}") + # custom format: relative-path (reject absolute glob patterns). if schema.get("format") == "relative-path" and value.startswith("/"): - errs.append(f"{path}: absolute paths are not allowed here (must be relative)") + errs.append( + f"{path}: absolute paths are not allowed here (must be relative)") if schema.get("type") == "integer": if "minimum" in schema and value < schema["minimum"]: diff --git a/agent-with-tools/agent.py b/agent-with-tools/agent.py index 35c51e2..3545629 100644 --- a/agent-with-tools/agent.py +++ b/agent-with-tools/agent.py @@ -78,7 +78,7 @@ def agent_loop(client): messages.append(message) if message.tool_calls: - # The LLM wants to use one or more tools — run them, then loop + # The LLM wants to use one or more tools, run them, then loop handle_tool_calls(message.tool_calls, messages) else: # No tool calls: we have the final answer