aws-samples
diff --git a/‎agent/src/hooks.py‎
Lines changed: 166 additions & 1 deletion b/‎agent/src/hooks.py‎
Lines changed: 166 additions & 1 deletion
diff --git a/‎agent/src/pipeline.py‎
Lines changed: 26 additions & 1 deletion b/‎agent/src/pipeline.py‎
Lines changed: 26 additions & 1 deletion
diff --git a/‎agent/src/policy.py‎
Lines changed: 24 additions & 2 deletions b/‎agent/src/policy.py‎
Lines changed: 24 additions & 2 deletions
@@ -99,6 +99,102 @@ def _tool_input_preview(tool_input: Any, max_len: int = TOOL_INPUT_PREVIEW_MAX)
     return _truncate(_strip_ansi(rendered), max_len)
 
 
+# ---------------------------------------------------------------------------
+# Egress-denial detection (#251, Phase 1)
+# ---------------------------------------------------------------------------
+#
+# Best-effort, heuristic scan of a tool's output for the signatures a blocked
+# outbound connection leaves behind — a non-allowlisted host hitting the DNS
+# Firewall, a refused connection, or a name-resolution failure. When one
+# matches we emit an ``egress_denied`` blocker naming the host to allowlist.
+# This is intentionally a string-signature scan (not a network probe): the
+# model's Bash tool calls (``npm install``, ``git clone``, ``curl``) surface
+# their stderr here, and those tools use stable, well-known phrasings. Patterns
+# that capture a host use a named ``host`` group; detection-only patterns omit
+# it (host stays None → the event still fires, just without ``resource``).
+_EGRESS_DENIAL_PATTERNS: tuple[re.Pattern[str], ...] = (
+    # curl / libcurl, git-over-https
+    re.compile(r"[Cc]ould not resolve host:?\s+(?P<host>[A-Za-z0-9._-]+)"),
+    re.compile(r"[Ff]ailed to connect to\s+(?P<host>[A-Za-z0-9._-]+)"),
+    # Node (npm/yarn), python (requests/urllib), getaddrinfo
+    re.compile(r"getaddrinfo (?:ENOTFOUND|EAI_AGAIN)\s+(?P<host>[A-Za-z0-9._-]+)"),
+    # requests/urllib3: the real host appears in ``HTTPSConnectionPool(host='x')``
+    # or after ``Failed to resolve 'x'``. Match those anchors specifically —
+    # a bare ``NameResolutionError.*?<fqdn>`` lazy-match would instead capture
+    # the ``urllib3.connection.HTTPSConnection`` class path that precedes the host.
+    re.compile(r"HTTPS?ConnectionPool\(host='(?P<host>[A-Za-z0-9._-]+)'"),
+    re.compile(r"Failed to resolve '(?P<host>[A-Za-z0-9._-]+)'"),
+    re.compile(
+        r"nodename nor servname provided|Temporary failure in name resolution"
+        r"|Name or service not known|Connection refused|Network is unreachable"
+        r"|NameResolutionError|Failed to establish a new connection",
+    ),
+)
+
+
+# #251 carry-path: the most recent agent-detected blocker, as a canonical
+# ``BLOCKED[<kind>]: …`` reason string (see ``format_blocker_reason``). Blockers
+# are detected mid-turn in the hooks (egress in PostToolUse, fail-closed in
+# PreToolUse) but the terminal ``error_message`` is assembled later in the
+# pipeline. This latch bridges the two: the pipeline promotes it into
+# ``TaskResult.error`` ONLY when the task errored without a more specific
+# reason, so the CDK classifier surfaces a precise remedy. Process-lifetime
+# (== one task) like ``_INJECTED_NUDGES``; last-writer-wins is fine — the most
+# recent blocker is the most relevant to a terminal failure.
+_LAST_BLOCKER_REASON: str | None = None
+
+
+def _record_blocker_reason(kind: str, detail: str, resource: str | None = None) -> None:
+    """Latch the canonical reason for the most recent agent-detected blocker."""
+    global _LAST_BLOCKER_REASON
+    from progress_writer import format_blocker_reason
+
+    _LAST_BLOCKER_REASON = format_blocker_reason(kind, detail, resource)
+
+
+def last_blocker_reason() -> str | None:
+    """Return the latched canonical blocker reason for terminal promotion (#251)."""
+    return _LAST_BLOCKER_REASON
+
+
+def reset_blocker_reason() -> None:
+    """Clear the in-process blocker latch.
+
+    Called at the start of each ``run_task`` (the latch is a process-scalar,
+    not task_id-keyed, so it must be reset per task to avoid leaking a prior
+    task's blocker) and by test fixtures for isolation.
+    """
+    global _LAST_BLOCKER_REASON
+    _LAST_BLOCKER_REASON = None
+
+
+# Back-compat alias for test fixtures that import the old name.
+_reset_blocker_reason_for_tests = reset_blocker_reason
+
+
+def detect_egress_denial(text: str) -> tuple[bool, str | None]:
+    """Scan tool output for an egress-denial signature (#251).
+
+    Returns ``(detected, host_or_None)``: ``(True, host)`` on the first matching
+    signature (``host`` populated when a pattern captured one so the emitted
+    ``egress_denied`` event can name the exact domain to allowlist; ``None`` for
+    detection-only signatures), or ``(False, None)`` when nothing matches.
+
+    Best-effort and deliberately conservative: it only fires on the
+    well-known phrasings the common CLIs (curl/git/npm/pip) emit, so a task
+    log that merely *mentions* "connection refused" in prose could produce a
+    false positive — acceptable for an observability signal that never blocks.
+    """
+    if not text:
+        return False, None
+    for pattern in _EGRESS_DENIAL_PATTERNS:
+        match = pattern.search(text)
+        if match:
+            host = match.groupdict().get("host") if match.groupdict() else None
+            return True, host
+    return False, None
+
+
 def _deny_response(reason: str) -> dict:
     """Build a PreToolUse DENY response with a sanitized reason.
 
@@ -214,6 +310,30 @@ async def pre_tool_use_hook(
                 "write_policy_decision_cached",
                 **decision.cache_hit_metadata,
             )
+        # #251 (decision E): a fail-closed deny means the Cedar engine errored
+        # or was unavailable — an ENVIRONMENTAL fault (misconfiguration), NOT
+        # an intentional hard-deny. Emit a ``policy_fail_closed`` blocker event
+        # so it renders distinctly and carries a remediation hint. We branch on
+        # the structured ``decision.fail_closed`` flag, never a reason-string
+        # prefix. Intentional hard-denies / cache-denies (fail_closed=False)
+        # never emit. Emitted strictly BEFORE the deny; the deny itself is
+        # unchanged, so this adds observability without altering the outcome.
+        if getattr(decision, "fail_closed", False):
+            _record_blocker_reason("policy_fail_closed", decision.reason)
+            if progress is not None:
+                _try_progress(
+                    progress,
+                    "write_agent_blocked",
+                    kind="policy_fail_closed",
+                    detail=decision.reason,
+                    remediation_hint=(
+                        "The Cedar policy engine errored or was unavailable "
+                        "(fail-closed). Check the agent policy-engine logs and the "
+                        "policy bundle; this is a misconfiguration, not an "
+                        "intentional deny."
+                    ),
+                    retryable=False,
+                )
         log("POLICY", f"DENIED: {tool_name} — {decision.reason}")
         return _deny_response(decision.reason)
 
@@ -918,13 +1038,19 @@ async def post_tool_use_hook(
     hook_context: Any,
     *,
     trajectory: _TrajectoryWriter | None = None,
+    progress: Any = None,
 ) -> dict:
     """PostToolUse hook: screen tool output for secrets/PII.
 
     Returns a dict with hookSpecificOutput.  When sensitive content is
     detected the response includes ``updatedMCPToolOutput`` containing the
     redacted version (steered enforcement — content is sanitized, not
     blocked).
+
+    ``progress`` is optional (preserves the Phase 1 test call shape). When
+    present, an egress-denial signature in the tool output emits an
+    ``egress_denied`` blocker event (#251) — best-effort observability,
+    never alters the screening decision.
     """
     _PASS_THROUGH: dict = {"hookSpecificOutput": {"hookEventName": "PostToolUse"}}
     _FAIL_CLOSED: dict = {
@@ -950,6 +1076,43 @@ async def post_tool_use_hook(
     if not isinstance(tool_response, str):
         tool_response = str(tool_response)
 
+    # #251: best-effort egress-denial detection. A blocked outbound connection
+    # (non-allowlisted host hitting the DNS Firewall, refused connection, name
+    # resolution failure) surfaces in the tool's stderr here. Emit an
+    # ``egress_denied`` blocker naming the host to allowlist. Observability
+    # only — never changes the pass-through / redaction decision below.
+    egress_detected, host = detect_egress_denial(tool_response)
+    if egress_detected:
+        detail = (
+            f"A connection to {host!r} was blocked"
+            if host
+            else "An outbound connection was blocked"
+        ) + f" during {tool_name}."
+        # Latch for terminal promotion ONLY when a host was captured. The latch
+        # feeds the authoritative TaskResult.error carry-path, so a host-less,
+        # detection-only signature ("Connection refused" / "Network is
+        # unreachable") — which commonly fires for an internal localhost race,
+        # not an egress denial — must not masquerade as the terminal cause. The
+        # observability event below still fires unconditionally (it never blocks
+        # and is clearly a heuristic signal in the live stream).
+        if host:
+            _record_blocker_reason("egress_denied", detail, host)
+        if progress is not None:
+            _try_progress(
+                progress,
+                "write_agent_blocked",
+                kind="egress_denied",
+                detail=detail,
+                remediation_hint=(
+                    f"Allowlist {host!r} in the DNS Firewall rule group"
+                    if host
+                    else "Allowlist the required host in the DNS Firewall rule group"
+                )
+                + " if it is a legitimate dependency. The agent never widens egress itself.",
+                retryable=False,
+                resource=host,
+            )
+
     try:
         result = scan_tool_output(tool_response)
     except Exception as exc:
@@ -1404,7 +1567,9 @@ async def _post(
         hook_input: HookInput, tool_use_id: str | None, ctx: HookContext
     ) -> HookJSONOutput:
         try:
-            result = await post_tool_use_hook(hook_input, tool_use_id, ctx, trajectory=trajectory)
+            result = await post_tool_use_hook(
+                hook_input, tool_use_id, ctx, trajectory=trajectory, progress=progress
+            )
             return SyncHookJSONOutput(**result)
         except Exception as exc:
             log("ERROR", f"PostToolUse wrapper crashed: {type(exc).__name__}: {exc}")
 
@@ -469,6 +469,16 @@ def _resolve_overall_task_status(
     if agent_status in ("success", "end_turn") and build_ok:
         return "success", err
 
+    # #251 carry-path: a hook may have detected an environmental blocker mid-run
+    # (egress denial, policy fail-closed) that the SDK surfaced only as a generic
+    # failure or as a missing ResultMessage. Promote the canonical
+    # ``BLOCKED[<kind>]: …`` reason so the CDK classifier attaches a precise
+    # remedy. Import locally to avoid a module-load cycle (hooks imports
+    # pipeline-adjacent modules).
+    from hooks import last_blocker_reason
+
+    blocker = last_blocker_reason()
+
     if agent_status == "unknown":
         if pr_url:
             log(
@@ -480,10 +490,17 @@ def _resolve_overall_task_status(
                 "INFO",
                 "No ResultMessage from SDK; build_ok=True (informational; task still failed)",
             )
+        # An egress denial that kills the agent's outbound calls is a likely
+        # cause of a missing ResultMessage — prefer the specific blocker reason
+        # over the generic SDK-no-result message when both are present.
+        if blocker and not err:
+            return "error", blocker
         merged = f"{err}; {_SDK_NO_RESULT_MESSAGE}" if err else _SDK_NO_RESULT_MESSAGE
         return "error", merged
 
     if not err:
+        if blocker:
+            return "error", blocker
         err = f"Task did not succeed (agent_status={agent_status!r}, build_ok={build_ok})"
     return "error", err
 
@@ -669,6 +686,14 @@ def run_task(
 
         agent_result: AgentResult | None = None
         progress = _ProgressWriter(config.task_id, trace=trace)
+        # #251: clear any blocker latched by a prior task. The agent container
+        # is one-task-per-process today, but the FastAPI server thread-pool can
+        # in principle dispatch a second run_task in the same process — reset
+        # here so a stale BLOCKED[...] reason can never leak into this task's
+        # terminal error_message (the latch is a scalar, not task_id-keyed).
+        from hooks import reset_blocker_reason
+
+        reset_blocker_reason()
         # --trace accumulator (design §10.1): when the task opted into
         # trace, ``_TrajectoryWriter`` keeps an in-memory copy of each
         # event so the pipeline can gzip+upload the full trajectory to
@@ -794,7 +819,7 @@ def _on_trace_truncated(max_bytes: int, first_dropped: int) -> None:
 
             # Setup repo (deterministic pre-hooks)
             with task_span("task.repo_setup") as setup_span:
-                setup = setup_repo(config)
+                setup = setup_repo(config, progress=progress)
                 setup_span.set_attribute("build.before", setup.build_before)
             progress.write_agent_milestone(
                 "repo_setup_complete",
 
@@ -205,6 +205,7 @@ class PolicyDecision:
     __slots__ = (
         "cache_hit_metadata",
         "duration_ms",
+        "fail_closed",
         "matching_rule_ids",
         "outcome",
         "reason",
@@ -223,6 +224,7 @@ def __init__(
         matching_rule_ids: tuple[str, ...] = (),
         duration_ms: float = 0.0,
         cache_hit_metadata: dict | None = None,
+        fail_closed: bool = False,
     ) -> None:
         if outcome is None and allowed is None:
             raise TypeError("PolicyDecision requires either outcome= or allowed=")
@@ -244,6 +246,16 @@ def __init__(
         # decisions with different original_decision_ts values still
         # represent the same deny outcome.
         self.cache_hit_metadata = cache_hit_metadata
+        # #251 (decision E): structured discriminator for environmental
+        # fail-closed denies (Cedar engine errored / unavailable) vs.
+        # intentional hard-denies and cache-driven denies. The hook branches
+        # on THIS flag — not a brittle ``reason.startswith("fail-closed:")``
+        # string match — to decide whether to emit a ``policy_fail_closed``
+        # blocker event. Set True ONLY at engine-error/unavailable deny sites;
+        # hard-deny and cache-deny leave it False. Like ``cache_hit_metadata``,
+        # NOT part of __eq__/__hash__ (it is derived from outcome+reason, and
+        # legacy equality-based tests predate the field).
+        self.fail_closed = fail_closed
 
     @property
     def allowed(self) -> bool:
@@ -296,8 +308,15 @@ def allow(cls, reason: str = "permitted", duration_ms: float = 0.0) -> PolicyDec
         return cls(outcome=Outcome.ALLOW, reason=reason, duration_ms=duration_ms)
 
     @classmethod
-    def deny(cls, reason: str, duration_ms: float = 0.0) -> PolicyDecision:
-        return cls(outcome=Outcome.DENY, reason=reason, duration_ms=duration_ms)
+    def deny(
+        cls, reason: str, duration_ms: float = 0.0, fail_closed: bool = False
+    ) -> PolicyDecision:
+        return cls(
+            outcome=Outcome.DENY,
+            reason=reason,
+            duration_ms=duration_ms,
+            fail_closed=fail_closed,
+        )
 
     @classmethod
     def require_approval(
@@ -1153,6 +1172,7 @@ def evaluate_tool_use(self, tool_name: str, tool_input: dict) -> PolicyDecision:
             return PolicyDecision.deny(
                 reason="policy engine unavailable",
                 duration_ms=(time.monotonic() - start) * 1000,
+                fail_closed=True,
             )
 
         base_context = {
@@ -1174,6 +1194,7 @@ def evaluate_tool_use(self, tool_name: str, tool_input: dict) -> PolicyDecision:
             return PolicyDecision.deny(
                 reason="fail-closed: unhashable_tool_input",
                 duration_ms=(time.monotonic() - start) * 1000,
+                fail_closed=True,
             )
 
         try:
@@ -1306,6 +1327,7 @@ def evaluate_tool_use(self, tool_name: str, tool_input: dict) -> PolicyDecision:
             return PolicyDecision.deny(
                 reason=f"fail-closed: {type(exc).__name__}",
                 duration_ms=(time.monotonic() - start) * 1000,
+                fail_closed=True,
             )
 
     # ---- Per-action routing ------------------------------------------------