fix(agent): address review findings — push -f gap, UserMessage handling, doc notes

- Add Cedar deny pattern for bare `git push -f` (no trailing args)
- Track UserMessage in message_counts, log string content
- Extract _format_tool_result helper to deduplicate ToolResultBlock formatting
- Parametrize quote-handling tests
- Document sentinel resource ID constraint for custom Cedar policies
  in policy.py module docstring and SECURITY.md

Author: bgagent

agent/src/policy.py

@@ -2,6 +2,27 @@
 
 Uses cedarpy (in-process Cedar evaluation) to enforce per-task-type
 tool restrictions. No network calls, no AWS Verified Permissions.
+
+Custom Cedar policies (via Blueprint ``security.cedarPolicies``) must
+use ``context`` conditions in ``when`` clauses — not ``resource ==``
+matching — for ``write_file`` and ``execute_bash`` actions.  The engine
+passes fixed sentinel resource IDs (``Agent::File::"file"``,
+``Agent::BashCommand::"command"``) because Cedar entity UIDs cannot
+contain the special characters found in file paths and bash commands.
+The actual values are available in ``context.file_path`` and
+``context.command`` respectively.  For ``invoke_tool`` actions, the
+resource ID is the real tool name (e.g. ``Agent::Tool::"Write"``), so
+``resource ==`` matching works normally there.
+
+Example — correct custom policy::
+
+    forbid (principal, action == Agent::Action::"execute_bash", resource)
+    when { context.command like "*curl*" };
+
+Example — WILL NOT WORK (resource is always ``"command"``)::
+
+    forbid (principal, action == Agent::Action::"execute_bash",
+        resource == Agent::BashCommand::"curl http://evil.com");
 """
 
 import time
@@ -39,6 +60,8 @@
 when { context.command like "*git push --force*" };
 forbid (principal, action == Agent::Action::"execute_bash", resource)
 when { context.command like "*git push -f *" };
+forbid (principal, action == Agent::Action::"execute_bash", resource)
+when { context.command like "*git push -f" };
 """
 
 
@@ -200,39 +223,32 @@ def evaluate_tool_use(self, tool_name: str, tool_input: dict) -> PolicyDecision:
                 )
                 return PolicyDecision(allowed=False, reason=reason, duration_ms=elapsed)
 
-            # Additional checks for Write/Edit: check file path.
-            # Use a fixed sentinel resource_id ("file") because file paths
-            # may contain quotes/special chars that break Cedar UID parsing.
-            # The actual path is in context.file_path where policies match it.
+            # Write/Edit: check file path against write_file policies.
+            # Sentinel resource_id avoids Cedar UID parsing issues (see _evaluate docstring).
             if tool_name in ("Write", "Edit"):
                 file_path = tool_input.get("file_path", "")
                 if file_path:
-                    file_context = {**base_context, "file_path": file_path}
                     allowed, deny_reason = self._evaluate(
                         "write_file",
                         "Agent::File",
                         "file",
-                        file_context,
+                        {**base_context, "file_path": file_path},
                     )
                     if not allowed:
                         elapsed = (time.monotonic() - start) * 1000
                         reason = deny_reason or f"Cedar policy denied write to {file_path}"
                         return PolicyDecision(allowed=False, reason=reason, duration_ms=elapsed)
 
-            # Additional checks for Bash: check command.
-            # Use a fixed sentinel resource_id ("command") because bash
-            # commands contain quotes/special chars that break Cedar UID
-            # parsing.  The actual command is in context.command where
-            # policies match it.
+            # Bash: check command against execute_bash policies.
+            # Sentinel resource_id avoids Cedar UID parsing issues (see _evaluate docstring).
             if tool_name == "Bash":
                 command = tool_input.get("command", "")
                 if command:
-                    bash_context = {**base_context, "command": command}
                     allowed, deny_reason = self._evaluate(
                         "execute_bash",
                         "Agent::BashCommand",
                         "command",
-                        bash_context,
+                        {**base_context, "command": command},
                     )
                     if not allowed:
                         elapsed = (time.monotonic() - start) * 1000

agent/src/runner.py

@@ -9,6 +9,13 @@
 from telemetry import _TrajectoryWriter
 
 
+def _format_tool_result(block) -> tuple[str, str]:
+    """Extract status label and content string from a ToolResultBlock."""
+    status = "ERROR" if block.is_error else "ok"
+    content = block.content if isinstance(block.content, str) else str(block.content)
+    return status, content
+
+
 def _setup_agent_env(config: dict) -> tuple[str | None, str | None]:
     """Configure process environment for the Claude Code CLI subprocess.
 
@@ -139,6 +146,7 @@ async def run_agent(
         ThinkingBlock,
         ToolResultBlock,
         ToolUseBlock,
+        UserMessage,
     )
 
     _setup_agent_env(config)
@@ -257,10 +265,7 @@ def _on_stderr(line: str) -> None:
                             log("TOOL", f"{block.name}: {truncate(str(tool_input))}")
                         turn_tool_calls.append({"name": block.name, "input": tool_input})
                     elif isinstance(block, ToolResultBlock):
-                        status = "ERROR" if block.is_error else "ok"
-                        content = (
-                            block.content if isinstance(block.content, str) else str(block.content)
-                        )
+                        status, content = _format_tool_result(block)
                         log("RESULT", f"[{status}] {truncate(content)}")
                         turn_tool_results.append(
                             {
@@ -352,6 +357,19 @@ def _on_stderr(line: str) -> None:
                     usage=usage_dict,
                 )
 
+            elif isinstance(message, UserMessage):
+                message_counts["other"] += 1
+                # UserMessage carries tool results fed back to the model.
+                # For hook-denied calls, content is a ToolResultBlock with
+                # is_error=True and the denial reason.
+                if isinstance(message.content, list):
+                    for block in message.content:
+                        if isinstance(block, ToolResultBlock):
+                            status, content = _format_tool_result(block)
+                            log("RESULT", f"[{status}] {truncate(content)}")
+                elif isinstance(message.content, str):
+                    log("USER", truncate(message.content))
+
             else:
                 message_counts["other"] += 1
                 log(

agent/tests/test_policy.py

@@ -123,6 +123,11 @@ def test_denies_git_push_f(self):
         result = engine.evaluate_tool_use("Bash", {"command": "git push -f origin main"})
         assert result.allowed is False
 
+    def test_denies_git_push_f_no_trailing_args(self):
+        engine = PolicyEngine(task_type="new_task", repo="owner/repo")
+        result = engine.evaluate_tool_use("Bash", {"command": "git push -f"})
+        assert result.allowed is False
+
     def test_allows_normal_bash(self):
         engine = PolicyEngine(task_type="new_task", repo="owner/repo")
         result = engine.evaluate_tool_use("Bash", {"command": "npm test"})
@@ -137,32 +142,19 @@ def test_allows_mise_run_build(self):
 class TestBashCommandsWithQuotes:
     """Commands containing double quotes must not cause NoDecision."""
 
-    def test_allows_git_commit_with_message(self):
-        engine = PolicyEngine(task_type="new_task", repo="owner/repo")
-        result = engine.evaluate_tool_use("Bash", {"command": 'git commit -m "fix: login bug"'})
-        assert result.allowed is True
-
-    def test_allows_git_commit_tree(self):
-        engine = PolicyEngine(task_type="new_task", repo="owner/repo")
-        cmd = 'git commit-tree HEAD^{tree} -m "squash"'
-        result = engine.evaluate_tool_use("Bash", {"command": cmd})
-        assert result.allowed is True
-
-    def test_allows_gh_pr_create(self):
-        engine = PolicyEngine(task_type="new_task", repo="owner/repo")
-        cmd = 'gh pr create --title "my PR" --body "desc"'
-        result = engine.evaluate_tool_use("Bash", {"command": cmd})
-        assert result.allowed is True
-
-    def test_allows_gh_api_post(self):
-        engine = PolicyEngine(task_type="new_task", repo="owner/repo")
-        cmd = 'gh api --method POST /repos/o/r/pulls -f title="PR"'
-        result = engine.evaluate_tool_use("Bash", {"command": cmd})
-        assert result.allowed is True
-
-    def test_allows_heredoc_commit(self):
+    @pytest.mark.parametrize(
+        "cmd",
+        [
+            'git commit -m "fix: login bug"',
+            'git commit-tree HEAD^{tree} -m "squash"',
+            'gh pr create --title "my PR" --body "desc"',
+            'gh api --method POST /repos/o/r/pulls -f title="PR"',
+            "git commit -m \"$(cat <<'EOF'\nFix the bug\nEOF\n)\"",
+        ],
+        ids=["git-commit-msg", "git-commit-tree", "gh-pr-create", "gh-api-post", "heredoc-commit"],
+    )
+    def test_allows_command_with_quotes(self, cmd):
         engine = PolicyEngine(task_type="new_task", repo="owner/repo")
-        cmd = "git commit -m \"$(cat <<'EOF'\nFix the bug\nEOF\n)\""
         result = engine.evaluate_tool_use("Bash", {"command": cmd})
         assert result.allowed is True
 

docs/design/SECURITY.md

@@ -257,7 +257,7 @@ AgentCore Memory has **no native backup mechanism**. This is a significant gap f
 - **No customer-managed KMS** — all encryption at rest uses AWS-managed keys. Customer-managed KMS can be added if required by compliance policy.
 - **CORS is fully open** — `ALL_ORIGINS` is configured for CLI consumption. Restrict origins when exposing browser clients.
 - **DNS Firewall IP bypass** — DNS Firewall does not block direct IP connections (see [NETWORK_ARCHITECTURE.md](./NETWORK_ARCHITECTURE.md#dns-firewall)).
-- **Partial tool access control** — Cedar-based policy enforcement (`agent/src/policy.py`) provides per-task-type tool restrictions (e.g. `pr_review` agents cannot use `Write`/`Edit`), path-based write protection, and destructive command blocking. Per-repo custom Cedar policies are supported via Blueprint `security.cedarPolicies`. Full tiered tool access (capability tiers, MCP server allowlisting) is planned for Iteration 5.
+- **Partial tool access control** — Cedar-based policy enforcement (`agent/src/policy.py`) provides per-task-type tool restrictions (e.g. `pr_review` agents cannot use `Write`/`Edit`), path-based write protection, and destructive command blocking. Per-repo custom Cedar policies are supported via Blueprint `security.cedarPolicies`. **Important:** custom policies for `write_file` and `execute_bash` actions must use `context.file_path` / `context.command` in `when` clauses — not `resource ==` matching — because the engine uses fixed sentinel resource IDs to avoid Cedar entity UID parsing failures on special characters. `invoke_tool` actions use the real tool name as resource ID, so `resource ==` matching works for tool-level policies. Full tiered tool access (capability tiers, MCP server allowlisting) is planned for Iteration 5.
 
 ## Reference