Merge branch 'main' into fix/502-ecs-payload-s3-pointer

Author: isadeks

.gitleaks.toml

@@ -19,6 +19,15 @@ stopwords = ["wat-opaque-123"]
 description = "Test fixture signing secret in Slack verification unit test (not a real credential)."
 stopwords = ["test-signing-secret-abc123"]
 
+[[allowlists]]
+# Test idempotency-key fixture in orchestration-release.test.ts (not a real
+# credential). Suppressed by stopword rather than a .gitleaksignore commit-SHA
+# fingerprint: the finding lives in history and SHA fingerprints break whenever
+# history is rewritten (rebases / dep-bump merges), which is exactly how #530's
+# baseline regressed. A stopword is SHA-independent. See #537 (regression of #530).
+description = "Test idempotency-key fixture 'orch_abc_SUB-1' (not a real credential)."
+stopwords = ["orch_abc_SUB-1"]
+
 # Catch bare 12-digit AWS account IDs. The default ruleset does not flag these,
 # which is how a real account ID reached a committed comment in the #236 integ
 # work. RE2 (Go) has no lookarounds, so the non-digit neighbours are captured in

.gitleaksignore

@@ -6,9 +6,7 @@
 0dca2171f244f693ac7c649ecf376d1ce6af2b38:docs/backlog/CLI_COGNITO_NEW_PASSWORD_CHALLENGE.md:aws-account-id:5
 0dca2171f244f693ac7c649ecf376d1ce6af2b38:docs/diagrams/phase3-cedar-hitl.drawio:aws-account-id:28
 
-# False positive: generic-api-key on a test idempotency-key fixture
-# ('orch_abc_SUB-1', not a real credential) in a file that no longer exists at
-# HEAD — it was removed when the #76 bounded-parallel perf commit was reverted
-# (e8bee5e). The finding lives only in history at f98f47a, so it can't be
-# scrubbed without rewriting shared `main`; fingerprint it instead. See #530.
-f98f47abc6f7820d7e5ce0375cf22551cb664b37:cdk/test/handlers/shared/orchestration-release.test.ts:generic-api-key:125
+# NOTE: the orch_abc_SUB-1 test-fixture false positive (formerly baselined here
+# by commit-SHA fingerprint, #530) is now suppressed SHA-independently via a
+# stopword allowlist in .gitleaks.toml — a commit-SHA fingerprint here broke when
+# history was rewritten. See #537.

agent/src/hooks.py

@@ -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}")

agent/src/observability.py

@@ -64,24 +64,54 @@ def current_otel_trace_id() -> str | None:
     bundle) so operators can join the task to its CloudWatch/X-Ray trace. Returns
     ``None`` when there is no recording span (e.g. tracing disabled locally) or
     the context is invalid, so callers can treat it as a graceful-missing field.
+
+    Never raises: a broken/misconfigured tracer degrades to ``None`` rather than
+    propagating. Callers read this inside DDB-write try-blocks (progress_writer),
+    where a raised trace error would otherwise be misclassified as a DDB failure
+    and trip the shared progress circuit breaker.
     """
-    span = trace.get_current_span()
-    ctx = span.get_span_context()
-    if not ctx.is_valid:
+    try:
+        span = trace.get_current_span()
+        ctx = span.get_span_context()
+        if not ctx.is_valid:
+            return None
+        # format_trace_id renders the 128-bit id as zero-padded 32-char hex — the
+        # OTEL format, so it joins directly in CloudWatch Transaction Search. Note
+        # the X-Ray console renders trace ids as ``1-{8hex}-{24hex}``; to look this
+        # up there, transform to that form (the timestamp is the first 8 hex chars).
+        return trace.format_trace_id(ctx.trace_id)
+    except Exception:
+        # nosemgrep: py-silent-success-masking -- trace id is a graceful-missing
+        # correlation field; a tracer fault must not fail the caller's write path.
         return None
-    # format_trace_id renders the 128-bit id as zero-padded 32-char hex — the
-    # OTEL format, so it joins directly in CloudWatch Transaction Search. Note
-    # the X-Ray console renders trace ids as ``1-{8hex}-{24hex}``; to look this
-    # up there, transform to that form (the timestamp is the first 8 hex chars).
-    return trace.format_trace_id(ctx.trace_id)
 
 
-def set_session_id(session_id: str) -> None:
-    """Propagate *session_id* via OTEL baggage for AgentCore session correlation.
+def propagate_correlation_context(
+    session_id: str,
+    user_id: str = "",
+    # ``user_id`` uses ""-means-absent (Cognito sub, mirrors AgentConfig.user_id
+    # which is never None); ``repo`` uses None-means-absent (mirrors the optional
+    # TaskRecord.repo). Both conventions flow from upstream config types; the
+    # ``if x:`` guards below flatten either to "don't set the baggage key".
+    repo: str | None = None,
+) -> None:
+    """Propagate the correlation envelope via OTEL baggage.
+
+    *session_id* correlates custom spans to the AgentCore session; *user_id*
+    and *repo* (#245) carry the platform identity and target repo so baggage
+    survives across pipeline phases on the task thread. Empty/None fields are
+    not set — so this runs (and is useful) even when *session_id* is empty but
+    the identity is known. *repo* is None for repo-less workflows (#248 Phase 3).
 
     The attached context is intentionally not detached: the background thread
     runs a single task then exits, so the context is garbage-collected with the
     thread.
     """
-    ctx = baggage.set_baggage("session.id", session_id)
+    ctx = context.get_current()
+    if session_id:
+        ctx = baggage.set_baggage("session.id", session_id, context=ctx)
+    if user_id:
+        ctx = baggage.set_baggage("user.id", user_id, context=ctx)
+    if repo:
+        ctx = baggage.set_baggage("repo.url", repo, context=ctx)
     context.attach(ctx)  # token not stored — thread-scoped lifetime

agent/src/pipeline.py

@@ -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
 
@@ -662,13 +679,26 @@ def run_task(
             "repo.url": config.repo_url,
             "issue.number": config.issue_number,
             "agent.model": config.anthropic_model,
+            # Correlation envelope (#245): user.id joins agent spans to
+            # orchestrator logs by the platform identity, not just task/repo.
+            **({"user.id": config.user_id} if config.user_id else {}),
         },
     ) as root_span:
         task_state.write_running(config.task_id)
         task_state.write_heartbeat(config.task_id)
 
         agent_result: AgentResult | None = None
-        progress = _ProgressWriter(config.task_id, trace=trace)
+        progress = _ProgressWriter(
+            config.task_id, trace=trace, user_id=config.user_id, repo=config.repo_url
+        )
+        # #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 +824,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",