pipeline.py

"""Task pipeline: the main orchestrator that wires all modules together."""

from __future__ import annotations

import asyncio
import hashlib
import os
import subprocess
import sys
import time
from typing import TYPE_CHECKING

from pydantic import ValidationError

import memory as agent_memory
import task_state
from channel_mcp import configure_channel_mcp
from config import AGENT_WORKSPACE, build_config, get_config, resolve_linear_api_token
from context import assemble_prompt, fetch_github_issue
from linear_reactions import react_task_finished, react_task_started
from models import AgentResult, HydratedContext, RepoSetup, TaskConfig, TaskResult
from observability import task_span
from post_hooks import (
    _extract_agent_notes,
    ensure_committed,
    ensure_pr,
    verify_build,
    verify_lint,
)
from progress_writer import _ProgressWriter
from prompt_builder import build_system_prompt, discover_project_config
from shell import log, log_error_cw
from system_prompt import SYSTEM_PROMPT
from telemetry import (
    _TrajectoryWriter,
    format_bytes,
    get_disk_usage,
    print_metrics,
    upload_trace_to_s3,
)

if TYPE_CHECKING:
    from workflow import Workflow

_SDK_NO_RESULT_MESSAGE = (
    "Agent SDK stream ended without a ResultMessage (agent_status=unknown). "
    "Treat as failure: possible SDK bug, network interruption, or protocol mismatch."
)


def _chain_prior_agent_error(agent_result: AgentResult | None, exc: BaseException) -> str:
    """Preserve agent-layer failures when a later pipeline stage raises."""
    tail = f"{type(exc).__name__}: {exc}"
    if agent_result is None:
        return tail
    if agent_result.error:
        return f"{agent_result.error}; subsequent failure: {tail}"
    if agent_result.status == "error":
        return f"Agent reported status=error; subsequent failure: {tail}"
    return tail


def _inject_attachment_context(prompt: str, prepared_attachments: list) -> str:
    """Append attachment file references to the user prompt.

    Images are referenced by absolute path so the agent can view them
    with the Read tool (which supports multimodal image reading).
    File attachments are similarly referenced by path.
    """
    lines = ["\n\n---\n\n**Attachments provided with this task:**\n"]
    for att in prepared_attachments:
        size_kb = att.size_bytes / 1024
        if att.type == "image":
            lines.append(
                f"- **Image:** `{att.filename}` ({size_kb:.1f} KB, {att.content_type}) "
                f"— View with: `Read {att.local_path}`"
            )
        else:
            lines.append(
                f"- **File:** `{att.filename}` ({size_kb:.1f} KB, {att.content_type}) "
                f"— Read with: `Read {att.local_path}`"
            )
    lines.append(
        "\nUse the Read tool to view these files. Image files will be displayed visually when read."
    )
    return prompt + "\n".join(lines)


def _maybe_upload_trace(
    config: TaskConfig,
    trajectory,
    progress,
) -> str | None:
    """Run the --trace S3 upload if the task opted in and user_id is set.

    Returns the resulting ``s3://`` URI (or ``None`` on any skip/fail).
    Fully fail-open: an exception here does NOT propagate. Called from
    both the happy path (post-hooks complete) and the crash path
    (top-level ``except``) so a crashing task still produces a
    debuggable artifact — which is exactly when ``--trace`` is most
    useful (K2 review Finding #1).

    Gates (K2 Stage 3 review Finding #1):
      - ``config.trace`` must be true.
      - ``config.user_id`` must be non-empty, else we would write to
        ``traces//<task_id>.jsonl.gz`` — an unreachable key that no
        Cognito caller can download through ``bgagent trace download``.
    """
    if not config.trace:
        return None
    if not config.user_id:
        log(
            "WARN",
            "Trace was enabled but user_id is empty — skipping S3 "
            "upload to avoid writing an unreachable artifact key. "
            f"task_id={config.task_id}",
        )
        return None
    try:
        artifact = trajectory.dump_gzipped_jsonl()
    except Exception as e:
        log("WARN", f"Trace dump_gzipped_jsonl failed: {type(e).__name__}: {e}")
        return None
    if not artifact:
        log(
            "INFO",
            "Trace accumulator is empty (no trajectory events captured). Skipping S3 upload.",
        )
        return None
    trace_s3_uri = upload_trace_to_s3(
        task_id=config.task_id,
        user_id=config.user_id,
        body=artifact,
    )
    if trace_s3_uri:
        try:
            progress.write_agent_milestone("trajectory_uploaded", trace_s3_uri)
        except Exception as e:
            # Milestone write is best-effort; don't mask the upload.
            log("WARN", f"trajectory_uploaded milestone emit failed: {type(e).__name__}: {e}")
        log("TASK", f"Trace artifact uploaded: {trace_s3_uri}")
    else:
        log(
            "WARN",
            "Trace upload returned no URI — see [trace/upload] logs "
            "above for the reason (skipped or failed). Task proceeds "
            "to terminal without trace_s3_uri.",
        )
    return trace_s3_uri


def _execute_agent_step(
    prompt: str,
    system_prompt: str,
    config,
    setup,
    hydrated,
    trajectory,
    progress,
):
    """Run the agentic step through the workflow step runner.

    Post-cutover (#248 task 8), the workflow runner is the sole path: the single
    ``run_agent`` step is dispatched through ``workflow.run_workflow`` —
    exercising the real handler registry, step milestones, and result threading —
    while clone, context assembly, and post-hooks stay inline (moving the full
    step list onto the runner is a follow-up). The workflow is loaded from the
    resolved ``{id, version}`` pinned at the create-task boundary.

    Returns the ``AgentResult`` so the surrounding pipeline (cancel short-circuit,
    post-hooks, result assembly) is unchanged.
    """
    from workflow import StepContext, load_workflow, run_workflow

    workflow_id = (config.resolved_workflow or {}).get("id", "coding/new-task-v1")
    wf = load_workflow(workflow_id)
    ctx = StepContext(
        workflow=wf,
        config=config,
        hydrated=hydrated,
        progress=progress,
        trajectory=trajectory,
        # Pre-populate the products the pipeline already built so the
        # clone_repo/hydrate_context handlers reuse them idempotently rather
        # than redo setup or rebuild the prompt the pipeline already injected
        # attachment context into.
        setup=setup,
        system_prompt=system_prompt,
        user_prompt=prompt,
    )
    # Drive only the agentic step through the runner; clone, context assembly,
    # and post-hooks stay on the inline path. only_kinds keeps the runner from
    # re-running the deterministic steps the pipeline already owns (double clone
    # / double PR).
    result = run_workflow(wf, ctx, only_kinds={"run_agent"})

    if ctx.agent_result is None:
        # The run_agent step did not produce a result — i.e. its handler raised
        # (run_workflow's _run_handler captures the exception into a failed
        # StepOutcome instead of propagating it). Re-raise here so run_task's
        # `except Exception` handles it with full fidelity: the log_error_cw
        # APPLICATION_LOGS mirror, the span error, and the real error text.
        detail = (
            result.failed_step.error
            if result.failed_step and result.failed_step.error
            else "run_agent step produced no result"
        )
        raise RuntimeError(f"Workflow run_agent step failed: {detail}")
    return ctx.agent_result


def _run_repoless_task(
    *,
    config,
    prompt: str,
    hc,
    progress,
    trajectory,
    root_span,
    start_time: float,
    memory_id: str,
    system_prompt_overrides: str,
) -> dict:
    """Run a repo-less workflow (#248 Phase 3) and return the result dict.

    No clone / build / PR: the workflow runner drives the full repo-less step
    list (``hydrate_context`` → ``run_agent`` → ``deliver_artifact``) inside the
    container, then a terminal ``TaskResult`` is assembled and persisted. The
    deliver_artifact step uploads the agent's result text to ``artifacts/{task_id}/``
    (and/or records a comment milestone), so the workflow's declared terminal
    outcome is actually produced; a delivery failure surfaces as a terminal
    FAILED rather than a silent "succeeded with nothing delivered".
    """
    from prompt_builder import build_repoless_system_prompt
    from workflow import StepContext, load_workflow, run_workflow

    workflow_id = (config.resolved_workflow or {}).get("id", "default/agent-v1")
    wf = load_workflow(workflow_id)
    system_prompt = build_repoless_system_prompt(config, hc, system_prompt_overrides)

    ctx = StepContext(
        workflow=wf,
        config=config,
        hydrated=hc,
        progress=progress,
        trajectory=trajectory,
        setup=None,  # repo-less: no RepoSetup
        system_prompt=system_prompt,
        user_prompt=prompt,
    )
    # Drive the full repo-less step list: hydrate_context → run_agent →
    # deliver_artifact. The deliverer uploads the agent's result text to
    # artifacts/{task_id}/ (and/or surfaces it as a comment), so the declared
    # terminal outcome is actually produced (#248 Phase 3).
    with task_span("task.agent_execution"):
        wf_result = run_workflow(wf, ctx)

    agent_result = ctx.agent_result
    if agent_result is None:
        # The run_agent step never produced a result — surface the underlying
        # failed-step error (e.g. the SDK loop raised) rather than a generic
        # message, so the terminal error is diagnosable.
        failed = wf_result.failed_step
        underlying = failed.error if failed and failed.error else None
        agent_result = AgentResult(
            status="error",
            error=(
                f"repo-less run_agent produced no result: {underlying}"
                if underlying
                else "repo-less run_agent produced no result"
            ),
        )
    progress.write_agent_milestone(
        "agent_execution_complete",
        f"status={agent_result.status} turns={agent_result.turns}",
    )

    duration = time.time() - start_time
    # No build/PR for a repo-less task; build_ok is vacuously true.
    overall_status, result_error = _resolve_overall_task_status(
        agent_result, build_ok=True, pr_url=None
    )

    # Delivery gate: deliver_artifact is the workflow's side-effecting terminal
    # step. WORKFLOWS.md defines primary:artifact success as "agent-success AND
    # an S3 artifact key is present" — so the gate has two arms:
    #   1. the runner reported a failed step (deliver_artifact raised), or
    #   2. the workflow's primary outcome is `artifact` but no artifact_uri was
    #      produced — i.e. delivery "succeeded" without writing the retrievable
    #      key the contract requires.
    # Either way the task produced nothing the user can retrieve, so it is a loud
    # FAILED rather than a silent "succeeded with no deliverable". Arm 2 closes
    # the gap where a deliverer that returns without raising (but also without an
    # S3 key) would otherwise pass the gate (code-review MEDIUM #1).
    artifact_uri = ctx.artifacts.get("artifact_uri")
    primary_outcome = wf.terminal_outcomes.primary
    if overall_status == "success" and not wf_result.succeeded:
        overall_status = "error"
        failed = wf_result.failed_step
        result_error = (
            f"Agent completed but delivery failed at step "
            f"{(failed.name if failed else 'deliver_artifact')!r}: "
            f"{(failed.error if failed and failed.error else 'unknown delivery error')}"
        )
        log("WARN", result_error)
    elif overall_status == "success" and primary_outcome == "artifact" and not artifact_uri:
        overall_status = "error"
        result_error = (
            "Agent completed and delivery reported success, but no artifact_uri "
            "was produced — the workflow's primary outcome is 'artifact', which "
            "requires a retrievable S3 key (WORKFLOWS.md success contract)."
        )
        log("WARN", result_error)

    trace_s3_uri = _maybe_upload_trace(config, trajectory, progress)

    # Episodic memory for a repo-less task is keyed on user:{user_id} (ADR-014
    # addendum) — the same namespace the orchestrator fallback + hydration read.
    # Fail-open: a memory write failure must not fail the task.
    memory_written = False
    effective_memory_id = memory_id or os.environ.get("MEMORY_ID", "")
    if effective_memory_id and config.user_id:
        import memory as agent_memory

        memory_written = agent_memory.write_task_episode(
            memory_id=effective_memory_id,
            actor=f"user:{config.user_id}",
            task_id=config.task_id,
            status="COMPLETED" if overall_status == "success" else "FAILED",
            cost_usd=agent_result.cost_usd,
            duration_s=round(duration, 1),
        )

    usage = agent_result.usage
    turns_attempted = agent_result.num_turns or agent_result.turns
    result = TaskResult(
        status=overall_status,
        agent_status=agent_result.status,
        pr_url=None,
        artifact_uri=artifact_uri,
        cost_usd=agent_result.cost_usd,
        turns=turns_attempted,
        turns_attempted=turns_attempted,
        turns_completed=_compute_turns_completed(
            agent_status=agent_result.status,
            turns_attempted=turns_attempted,
            max_turns=config.max_turns,
        ),
        duration_s=round(duration, 1),
        task_id=config.task_id,
        memory_written=memory_written,
        error=result_error,
        session_id=agent_result.session_id or None,
        input_tokens=usage.input_tokens if usage else None,
        output_tokens=usage.output_tokens if usage else None,
        cache_read_input_tokens=usage.cache_read_input_tokens if usage else None,
        cache_creation_input_tokens=usage.cache_creation_input_tokens if usage else None,
        trace_s3_uri=trace_s3_uri,
    )
    result_dict = result.model_dump()

    root_span.set_attribute("task.status", result.status)
    root_span.set_attribute("task.repo_less", True)
    if result.cost_usd is not None:
        root_span.set_attribute("agent.cost_usd", float(result.cost_usd))

    print_metrics(result_dict)
    terminal_status = "COMPLETED" if overall_status == "success" else "FAILED"
    task_state.write_terminal(config.task_id, terminal_status, result_dict)
    return result_dict


def _apply_post_hook_gates(
    workflow: Workflow | None,
    *,
    read_only: bool,
    build_passed: bool,
    lint_passed: bool,
    build_before: bool,
    lint_before: bool,
) -> bool:
    """Resolve the coding lane's post-hook verify gates against the workflow (#301).

    Decision (issue #301 acceptance criteria): the inline post-hook path
    CONSULTS each declared ``verify_build`` / ``verify_lint`` step's ``gate``
    through the runner's ``gate_status`` — the single place gate semantics live —
    rather than routing the post-hooks through the runner's step handlers.
    Routing through the runner would also change failure-path side effects (a
    gating ``verify_build`` with ``on_failure: fail`` stops the runner *before*
    ``ensure_pr``, stranding committed work with no PR), which is the broader
    half-migrated-runner unification the issue defers. Here the inline ordering
    (verify → ensure_pr always runs) is preserved; only the task verdict honors
    the declared gate.

    Per-step semantics:

    - A declared step gates per its ``gate`` (``strict`` | ``regression_only`` |
      ``informational``; unset = ``regression_only``), but only when its
      ``on_failure`` is ``fail`` — ``continue``/``skip_remaining`` steps are
      advisory for the task verdict, matching the runner.
    - An undeclared ``verify_build`` keeps the legacy regression-only gating
      (identical to ``gate_status`` with ``gate=None``).
    - An undeclared ``verify_lint`` never gates (legacy: lint is not used for
      terminal status unless a workflow opts in by declaring the step).
    - ``workflow is None`` (post-hook reload failed) falls back to the legacy
      gating for both, so a corrupt file cannot strand the agent's work.
    """
    from workflow import gate_status

    steps = list(workflow.steps) if workflow is not None else []
    gates_ok = True
    for kind, passed, was_passing_before in (
        ("verify_build", build_passed, build_before),
        ("verify_lint", lint_passed, lint_before),
    ):
        step = next((s for s in steps if s.kind == kind), None)
        if step is None:
            if kind == "verify_lint":
                continue
            gate, gating, on_failure = None, True, "fail"
        else:
            gate, gating, on_failure = step.gate, step.on_failure == "fail", step.on_failure
        status = gate_status(
            passed=passed,
            gate=gate,
            read_only=read_only,
            was_passing_before=was_passing_before,
        )
        if passed:
            continue
        label = gate or "regression_only"
        if status == "succeeded":
            if read_only:
                log("INFO", f"read-only workflow: {kind} failed — informational only, not gating")
            elif gate == "informational":
                log("INFO", f"{kind} failed — gate=informational, not gating")
            else:
                log(
                    "WARN",
                    f"Post-agent {kind} failed, but it was already failing before "
                    "agent changes — not counting as regression",
                )
        elif gating:
            log("WARN", f"{kind} failed — gate={label} gates the task")
            gates_ok = False
        else:
            log("INFO", f"{kind} failed — gate={label} but on_failure={on_failure}, not gating")
    return gates_ok


def _resolve_overall_task_status(
    agent_result: AgentResult,
    *,
    build_ok: bool,
    pr_url: str | None,
) -> tuple[str, str | None]:
    """Map agent outcome + build gate to (overall_status, error_for_task_result)."""
    agent_status = agent_result.status
    err = agent_result.error

    if agent_status in ("success", "end_turn") and build_ok:
        return "success", err

    if agent_status == "unknown":
        if pr_url:
            log(
                "INFO",
                f"No ResultMessage from SDK (agent_status=unknown); pr_url present: {pr_url}",
            )
        if build_ok:
            log(
                "INFO",
                "No ResultMessage from SDK; build_ok=True (informational; task still failed)",
            )
        merged = f"{err}; {_SDK_NO_RESULT_MESSAGE}" if err else _SDK_NO_RESULT_MESSAGE
        return "error", merged

    if not err:
        err = f"Task did not succeed (agent_status={agent_status!r}, build_ok={build_ok})"
    return "error", err


def _compute_turns_completed(
    agent_status: str,
    turns_attempted: int | None,
    max_turns: int,
) -> int | None:
    """Clamp ``turns_completed`` to ``max_turns`` when the SDK hit the limit.

    Rev-5 DATA-1 — the Claude Agent SDK reports ``num_turns = max_turns + 1``
    on ``error_max_turns`` because the aborted attempt is counted.  Clamping
    at the final write keeps ``turns_completed`` truthful ("how many turns
    actually executed") while ``turns_attempted`` keeps the raw SDK value
    for debugging.

    Returns ``None`` if ``turns_attempted`` is ``None``/falsy so callers can
    round-trip a missing SDK count without inventing a fake zero.
    """
    if not turns_attempted:
        return turns_attempted
    if agent_status == "error_max_turns":
        return min(turns_attempted, max_turns)
    return turns_attempted


def _write_memory(
    config: TaskConfig,
    setup: RepoSetup,
    agent_result: AgentResult,
    start_time: float,
    build_passed: bool,
    pr_url: str | None,
    memory_id: str,
) -> bool:
    """Write task episode and repo learnings to AgentCore Memory.

    Returns True if any memory was successfully written.
    """
    # Parse self-feedback from PR body — separate try-catch so extraction
    # failures don't mask memory write errors (and vice versa).
    self_feedback = None
    try:
        self_feedback = _extract_agent_notes(setup.repo_dir, setup.branch, config)
    except Exception as e:
        log(
            "WARN",
            f"Agent notes extraction failed (non-fatal): {type(e).__name__}: {e}",
        )

    episode_cost = agent_result.cost_usd

    # Memory writes are individually fail-open (return False on error)
    episode_ok = agent_memory.write_task_episode(
        memory_id=memory_id,
        actor=config.repo_url,
        task_id=config.task_id,
        status="COMPLETED" if build_passed else "FAILED",
        pr_url=pr_url,
        cost_usd=episode_cost,
        duration_s=round(time.time() - start_time, 1),
        self_feedback=self_feedback,
    )

    learnings_ok = False
    if self_feedback:
        learnings_ok = agent_memory.write_repo_learnings(
            memory_id=memory_id,
            repo=config.repo_url,
            task_id=config.task_id,
            learnings=self_feedback,
        )

    log("MEMORY", f"Memory write: episode={episode_ok}, learnings={learnings_ok}")
    return episode_ok or learnings_ok


def run_task(
    repo_url: str = "",
    task_description: str = "",
    issue_number: str = "",
    github_token: str = "",
    anthropic_model: str = "",
    max_turns: int = 100,
    max_budget_usd: float | None = None,
    aws_region: str = "",
    task_id: str = "",
    hydrated_context: dict | None = None,
    system_prompt_overrides: str = "",
    prompt_version: str = "",
    memory_id: str = "",
    resolved_workflow: dict | None = None,
    branch_name: str = "",
    pr_number: str = "",
    cedar_policies: list[str] | None = None,
    approval_timeout_s: int | None = None,
    initial_approvals: list[str] | None = None,
    initial_approval_gate_count: int = 0,
    approval_gate_cap: int | None = None,
    channel_source: str = "",
    channel_metadata: dict[str, str] | None = None,
    trace: bool = False,
    user_id: str = "",
    attachments: list[dict] | None = None,
) -> dict:
    """Run the full agent pipeline and return a serialized result dict.

    This is the main entry point for both:
      - AgentCore server mode (called by server.py /invocations)
      - Local batch mode (called by main())

    Builds a ``TaskResult`` Pydantic model internally, then returns
    ``TaskResult.model_dump()`` for downstream consumers (DynamoDB,
    metrics, server response).
    """
    from opentelemetry.trace import StatusCode

    from repo import setup_repo

    # Build config
    config = build_config(
        repo_url=repo_url,
        task_description=task_description,
        issue_number=issue_number,
        github_token=github_token,
        anthropic_model=anthropic_model,
        max_turns=max_turns,
        max_budget_usd=max_budget_usd,
        aws_region=aws_region,
        task_id=task_id,
        system_prompt_overrides=system_prompt_overrides,
        resolved_workflow=resolved_workflow,
        branch_name=branch_name,
        pr_number=pr_number,
        channel_source=channel_source,
        channel_metadata=channel_metadata,
        trace=trace,
        user_id=user_id,
        approval_timeout_s=approval_timeout_s,
        initial_approvals=initial_approvals,
        initial_approval_gate_count=initial_approval_gate_count,
        approval_gate_cap=approval_gate_cap,
        attachments=attachments,
    )

    # Inject Cedar policies into config for the PolicyEngine in runner.py
    if cedar_policies:
        config.cedar_policies = cedar_policies

    # Export session-tag values so tenant-data boto3 clients (DDB/S3) assume
    # the per-task SessionRole with {user_id, repo, task_id} tags. No-op when
    # AGENT_SESSION_ROLE_ARN is unset (local/dev/tests).
    from aws_session import configure_session, is_scoped

    configure_session(
        user_id=config.user_id,
        repo=config.repo_url,
        task_id=config.task_id,
    )
    # Surface the credential-scoping posture once per task so every task's logs
    # state plainly whether tenant-data isolation was active. is_scoped()
    # resolves the session; if scoping was requested but unbuildable it raises
    # SessionScopingError here (fail closed) rather than running unscoped.
    log("TASK", f"Tenant-data credential scoping: {'SCOPED' if is_scoped() else 'UNSCOPED'}")

    log("TASK", f"Task ID: {config.task_id}")
    log("TASK", f"Repository: {config.repo_url}")
    log("TASK", f"Issue: {config.issue_number or '(none)'}")
    log("TASK", f"Model: {config.anthropic_model}")

    with task_span(
        "task.pipeline",
        attributes={
            "task.id": config.task_id,
            "repo.url": config.repo_url,
            "issue.number": config.issue_number,
            "agent.model": config.anthropic_model,
        },
    ) 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)
        # --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
        # S3 on terminal. Owned by the pipeline rather than the runner
        # so the accumulator outlives ``run_agent``'s scope.
        trajectory = _TrajectoryWriter(config.task_id, accumulate=trace)
        # K2 review Finding #3 — surface accumulator truncation to the
        # user via a ``trace_truncated`` milestone on TaskEventsTable
        # (visible in ``bgagent watch``). Fire-once by design: the
        # downloaded artifact's header reports the final drop count.
        if trace:

            def _on_trace_truncated(max_bytes: int, first_dropped: int) -> None:
                progress.write_agent_milestone(
                    "trace_truncated",
                    f"Trace accumulator hit its {max_bytes}-byte cap after "
                    f"{first_dropped} event drop(s); the downloaded "
                    f"artifact will be truncated. See the "
                    f"TRAJECTORY_ARTIFACT_HEADER row for the final "
                    f"drop count.",
                )

            trajectory.set_truncation_callback(_on_trace_truncated)
        # Declared up-front so the crash handler at the bottom of this `try`
        # can reference it via a normal name rather than ``locals().get(...)``
        # — survives refactors and reads cleanly. Stays None until the Linear
        # `react_task_started` call assigns the actual reaction id.
        linear_eyes_reaction_id: str | None = None
        try:
            # Context hydration
            with task_span("task.context_hydration"):
                if hydrated_context:
                    log("TASK", "Using hydrated context from orchestrator")
                    try:
                        hc = HydratedContext.model_validate(hydrated_context)
                    except ValidationError as err:
                        parts = [
                            f"{'.'.join(str(x) for x in e['loc'])}: {e['msg']}"
                            for e in err.errors()
                        ]
                        log(
                            "ERROR",
                            "HydratedContext validation failed (orchestrator vs agent contract): "
                            + "; ".join(parts),
                        )
                        raise
                    prompt = hc.user_prompt
                    if hc.issue:
                        config.issue = hc.issue
                    if hc.resolved_branch_name:
                        config.branch_name = hc.resolved_branch_name
                    if hc.resolved_base_branch:
                        config.base_branch = hc.resolved_base_branch
                    if hc.truncated:
                        log("WARN", "Context was truncated by orchestrator token budget")
                    if hc.fallback_error:
                        log("WARN", f"Orchestrator context fallback: {hc.fallback_error}")
                    if hc.guardrail_blocked:
                        log(
                            "WARN",
                            f"Orchestrator guardrail blocked content: {hc.guardrail_blocked}",
                        )
                else:
                    hc = None
                    # Local batch mode — fetch issue and assemble prompt in-container
                    if config.issue_number:
                        log("TASK", f"Fetching issue #{config.issue_number}...")
                        config.issue = fetch_github_issue(
                            config.repo_url, config.issue_number, config.github_token
                        )
                        log("TASK", f"  Title: {config.issue.title}")

                    prompt = assemble_prompt(config)

            # Repo-less path (#248 Phase 3): a knowledge workflow has no repo to
            # clone, build, or PR. Drive its steps (hydrate_context → run_agent →
            # deliver_artifact) through the workflow runner and assemble the
            # terminal result, skipping the repo-coupled segment below entirely.
            #
            # ``requires_repo: false`` means repo-OPTIONAL, not repo-forbidden:
            # create-task-core admits and persists a repo for such a workflow,
            # and the orchestrator then assembles a repo-bound prompt (issue/PR
            # fetch). Keying the repo-less branch on ``requires_repo`` ALONE made
            # the agent skip the clone while the prompt promised a repo — the two
            # halves disagreed (PR review #296 finding #3). So take the repo-less
            # path only when no repo was actually supplied; when a repo IS present
            # it is hydrated as context exactly like a coding task (clone → build →
            # PR), honoring the repo-bound prompt the orchestrator built.
            if not config.requires_repo and not config.repo_url:
                return _run_repoless_task(
                    config=config,
                    prompt=prompt,
                    hc=hc,
                    progress=progress,
                    trajectory=trajectory,
                    root_span=root_span,
                    start_time=time.time(),
                    memory_id=memory_id,
                    system_prompt_overrides=system_prompt_overrides,
                )

            # Configure git and gh auth before setup_repo() uses them
            subprocess.run(
                ["git", "config", "--global", "user.name", "bgagent"],
                check=True,
                capture_output=True,
                timeout=60,
            )
            subprocess.run(
                ["git", "config", "--global", "user.email", "bgagent@noreply.github.com"],
                check=True,
                capture_output=True,
                timeout=60,
            )
            os.environ["GITHUB_TOKEN"] = config.github_token
            os.environ["GH_TOKEN"] = config.github_token

            # Set env vars for the prepare-commit-msg hook BEFORE setup_repo()
            # so the hook has access to TASK_ID/PROMPT_VERSION from the start.
            os.environ["TASK_ID"] = config.task_id
            if prompt_version:
                os.environ["PROMPT_VERSION"] = prompt_version

            # Setup repo (deterministic pre-hooks)
            with task_span("task.repo_setup") as setup_span:
                setup = setup_repo(config)
                setup_span.set_attribute("build.before", setup.build_before)
            progress.write_agent_milestone(
                "repo_setup_complete",
                f"branch={setup.branch} build_before={setup.build_before}",
            )

            system_prompt = build_system_prompt(config, setup, hc, system_prompt_overrides)

            # Channel-specific MCP wiring (Linear only, for v1). Must happen
            # before discover_project_config so the scan picks up the file we
            # just wrote. Resolve the API token from Secrets Manager *before*
            # writing .mcp.json so the child SDK process inherits the env var
            # that the MCP server entry references via ${LINEAR_API_TOKEN}.
            if config.channel_source == "linear":
                resolve_linear_api_token(config.channel_metadata)
            configure_channel_mcp(setup.repo_dir, config.channel_source)

            # 👀 on the Linear issue — acknowledges the task is picked up.
            # No-op for non-Linear tasks. Best-effort; failures are logged
            # but do not block the pipeline. Capture the reaction id so we
            # can delete it at terminal status (👀 → ✅/❌).
            linear_eyes_reaction_id = react_task_started(
                config.channel_source,
                config.channel_metadata,
            )

            # Download attachments from S3 (version-pinned, integrity-verified)
            prepared_attachments: list = []
            if config.attachments:
                from attachments import download_attachments

                try:
                    with task_span("task.attachment_download"):
                        prepared_attachments = download_attachments(
                            config.attachments, setup.repo_dir
                        )
                    progress.write_agent_milestone(
                        "attachments_downloaded",
                        f"count={len(prepared_attachments)}",
                    )
                except RuntimeError as e:
                    log("ERROR", f"Attachment integrity check failed: {e}")
                    raise RuntimeError(
                        f"Attachment download/verification failed: {e}. "
                        "The task cannot proceed without valid attachments."
                    ) from e
                except Exception as e:
                    err_type = type(e).__name__
                    log("ERROR", f"Attachment download failed: {err_type}: {e}")
                    raise RuntimeError(
                        f"Failed to download task attachments from S3: {err_type}: {e}"
                    ) from e

            # Log discovered repo-level project configuration
            # (all files loaded by setting_sources=["project"])
            repo_dir = setup.repo_dir
            project_config = discover_project_config(repo_dir)
            if project_config:
                log("TASK", f"Repo project configuration: {project_config}")
            else:
                log("TASK", "No repo-level project configuration found")

            # Inject attachment references into the prompt so the agent knows
            # about available files. Images are read natively by the agent's
            # Read tool (multimodal support). File attachments are referenced
            # by path for the agent to read as needed.
            if prepared_attachments:
                prompt = _inject_attachment_context(prompt, prepared_attachments)

            # Run agent
            disk_before = get_disk_usage(AGENT_WORKSPACE)
            start_time = time.time()

            log("TASK", "Starting agent...")
            if config.max_budget_usd:
                log("TASK", f"Budget limit: ${config.max_budget_usd:.2f}")
            # Warn if uvloop is the active policy — subprocess SIGCHLD conflicts.
            policy = asyncio.get_event_loop_policy()
            policy_name = type(policy).__name__
            if "uvloop" in policy_name.lower():
                log(
                    "WARN",
                    f"uvloop detected ({policy_name}) — this may cause subprocess "
                    f"SIGCHLD conflicts with the Claude Agent SDK",
                )
            with task_span("task.agent_execution") as agent_span:
                try:
                    agent_result = _execute_agent_step(
                        prompt,
                        system_prompt,
                        config,
                        setup,
                        hc,
                        trajectory,
                        progress,
                    )
                except Exception as e:
                    # Fatal agent error: mirror to APPLICATION_LOGS so
                    # TaskDashboard widgets + ``bgagent status`` can see
                    # the real failure text instead of stopping at
                    # ``error_classification.UNKNOWN``. Local stdout
                    # path is preserved for docker-compose / unit-test
                    # capture.
                    log_error_cw(f"Agent failed: {e}", task_id=config.task_id or None)
                    agent_span.set_status(StatusCode.ERROR, str(e))
                    agent_span.record_exception(e)
                    agent_result = AgentResult(status="error", error=str(e))
            progress.write_agent_milestone(
                "agent_execution_complete",
                f"status={agent_result.status} turns={agent_result.turns}",
            )

            # Cancel short-circuit: the Stop hook signalled cancel by stopping
            # the SDK early, but that only stops the agent loop — post-hooks
            # (ensure_committed, ensure_pr) would still run and push/open a PR
            # on a cancelled task.  Re-check the task status here and exit the
            # pipeline before any side-effect-producing post-hook runs.  The
            # terminal state is already CANCELLED (written by cancel-task.ts),
            # so we do NOT call write_terminal — its ConditionExpression only
            # allows RUNNING/HYDRATING/FINALIZING, which would fail silently,
            # but leaving the cancel record intact makes the intent explicit.
            try:
                _current_record = task_state.get_task(config.task_id)
            except task_state.TaskFetchError:
                _current_record = None  # fail-open: let normal path proceed
            if _current_record and _current_record.get("status") == "CANCELLED":
                log("TASK", f"Task {config.task_id} cancelled; skipping post-hooks")
                progress.write_agent_milestone(
                    "task_cancelled_acknowledged",
                    "Post-hooks skipped; terminal state already CANCELLED.",
                )
                # L4 item 1c: best-effort trace upload + conditional
                # self-heal on the cancel path. ``write_terminal``'s
                # ConditionExpression rejects CANCELLED, so we cannot
                # persist ``trace_s3_uri`` atomically with the terminal
                # write — use ``write_trace_uri_conditional`` instead,
                # which is scoped to ``attribute_not_exists(trace_s3_uri)``
                # AND a terminal status. Fully fail-open: any exception
                # (upload, DDB, serialization) must not prevent the
                # cancel fast-path from returning.
                if config.trace:
                    log(
                        "TASK",
                        "Task cancelled mid-run; attempting best-effort "
                        "--trace upload + conditional persist so the "
                        "trajectory captured before cancel is still "
                        "recoverable.",
                    )
                    try:
                        trace_s3_uri = _maybe_upload_trace(config, trajectory, progress)
                        if trace_s3_uri:
                            task_state.write_trace_uri_conditional(config.task_id, trace_s3_uri)
                    except Exception as e:
                        log(
                            "WARN",
                            f"Cancel-path trace upload/persist failed "
                            f"(fail-open): {type(e).__name__}: {e}",
                        )
                return {
                    "status": "cancelled",
                    "task_id": config.task_id,
                    "turns": agent_result.turns,
                    "turns_attempted": agent_result.num_turns or agent_result.turns,
                }

            # Resolve the post-hook gating inputs: read_only, the ensure_pr
            # strategy (create / push_resolve / resolve), and the verify steps'
            # declared gates (#301) the workflow declares.
            #
            # ``read_only`` comes from ``config`` — build_config already computed
            # it (with its own fail-soft fallback) and it drove Cedar during the
            # run, so reusing it keeps the post-hook on the SAME verdict rather
            # than re-deriving a possibly-divergent one. The workflow file is
            # reloaded for the ensure_pr STRATEGY and the verify-step GATES, and
            # that reload is wrapped in the same WorkflowValidationError fallback
            # build_config uses (config.py): this code path runs AFTER run_agent
            # has already mutated / committed the tree, so a load failure here
            # must NOT strand the work as FAILED with no PR — it falls back to
            # the default "create" strategy + legacy regression-only gating and
            # still opens the PR (PR review #296 finding #5).
            from workflow import WorkflowValidationError, load_workflow

            workflow_read_only = config.read_only
            _workflow = None
            try:
                _workflow = load_workflow(
                    (config.resolved_workflow or {}).get("id", "coding/new-task-v1")
                )
                ensure_pr_strategy = next(
                    (s.strategy for s in _workflow.steps if s.kind == "ensure_pr" and s.strategy),
                    "create",
                )
            except WorkflowValidationError as exc:
                log(
                    "WARN",
                    f"post-hook workflow reload failed ({exc}); defaulting ensure_pr "
                    "strategy to 'create' so the agent's work is not stranded",
                )
                ensure_pr_strategy = "create"

            # Post-hooks (agent_result is guaranteed set by the try/except above)
            with task_span("task.post_hooks") as post_span:
                # Safety net: commit any uncommitted tracked changes (skip for read-only tasks)
                safety_committed = False if workflow_read_only else ensure_committed(setup.repo_dir)
                post_span.set_attribute("safety_net.committed", safety_committed)

                build_passed = verify_build(setup.repo_dir)
                lint_passed = verify_lint(setup.repo_dir)
                pr_url = ensure_pr(
                    config,
                    setup,
                    build_passed,
                    lint_passed,
                    agent_result=agent_result,
                    strategy=ensure_pr_strategy,
                )
                post_span.set_attribute("build.passed", build_passed)
                post_span.set_attribute("lint.passed", lint_passed)
                post_span.set_attribute("pr.url", pr_url or "")
            if pr_url:
                progress.write_agent_milestone("pr_created", pr_url)

            # Memory write — capture task episode and repo learnings
            memory_written = False
            effective_memory_id = memory_id or os.environ.get("MEMORY_ID", "")
            if effective_memory_id:
                memory_written = _write_memory(
                    config,
                    setup,
                    agent_result,
                    start_time,
                    build_passed,
                    pr_url,
                    effective_memory_id,
                )

            # Metrics
            duration = time.time() - start_time
            disk_after = get_disk_usage(AGENT_WORKSPACE)

            # Overall status: do not infer success from PR/build when the SDK never
            # emitted ResultMessage (agent_status=unknown) — that masks protocol gaps.
            # Gating honors each verify step's declared ``gate`` via the runner's
            # gate_status (#301); an undeclared verify_lint never gates (legacy).
            agent_status = agent_result.status
            build_ok = _apply_post_hook_gates(
                _workflow,
                read_only=workflow_read_only,
                build_passed=build_passed,
                lint_passed=lint_passed,
                # setup defaults assume green-before, so a post-agent failure IS
                # counted as a regression (conservative).
                build_before=setup.build_before,
                lint_before=setup.lint_before,
            )
            overall_status, result_error = _resolve_overall_task_status(
                agent_result,
                build_ok=build_ok,
                pr_url=pr_url,
            )

            # ✅/❌ on the Linear issue (removes the 👀 first so the final
            # status stands alone). No-op for non-Linear tasks.
            react_task_finished(
                config.channel_source,
                config.channel_metadata,
                success=(overall_status == "success"),
                started_reaction_id=linear_eyes_reaction_id,
            )

            # --trace trajectory S3 upload (design §10.1). Runs AFTER
            # post-hooks but BEFORE ``write_terminal`` so the resulting
            # ``trace_s3_uri`` can be persisted atomically with the
            # terminal-status transition. Fail-open: an S3 error does
            # NOT flip the task to FAILED — the trajectory is a debug
            # artifact, not a correctness gate. The same helper is also
            # invoked from the crash path below so a pipeline exception
            # still produces a usable debug artifact.
            trace_s3_uri = _maybe_upload_trace(config, trajectory, progress)

            # Build TaskResult
            usage = agent_result.usage
            turns_attempted = agent_result.num_turns or agent_result.turns
            turns_completed = _compute_turns_completed(
                agent_status=agent_status,
                turns_attempted=turns_attempted,
                max_turns=config.max_turns,
            )
            result = TaskResult(
                status=overall_status,
                agent_status=agent_status,
                pr_url=pr_url,
                build_passed=build_passed,
                lint_passed=lint_passed,
                cost_usd=agent_result.cost_usd,
                # Legacy field (= turns_attempted) kept for back-compat.
                turns=turns_attempted,
                turns_attempted=turns_attempted,
                turns_completed=turns_completed,
                duration_s=round(duration, 1),
                task_id=config.task_id,
                disk_before=format_bytes(disk_before),
                disk_after=format_bytes(disk_after),
                disk_delta=format_bytes(disk_after - disk_before),
                prompt_version=prompt_version or None,
                memory_written=memory_written,
                error=result_error,
                session_id=agent_result.session_id or None,
                input_tokens=usage.input_tokens if usage else None,
                output_tokens=usage.output_tokens if usage else None,
                cache_read_input_tokens=usage.cache_read_input_tokens if usage else None,
                cache_creation_input_tokens=usage.cache_creation_input_tokens if usage else None,
                trace_s3_uri=trace_s3_uri,
            )

            result_dict = result.model_dump()

            # Record terminal attributes on the root span for CloudWatch querying
            root_span.set_attribute("task.status", result.status)
            if result.cost_usd is not None:
                root_span.set_attribute("agent.cost_usd", float(result.cost_usd))
            if result.turns:
                root_span.set_attribute("agent.turns", int(result.turns))
            root_span.set_attribute("build.passed", result.build_passed)
            root_span.set_attribute("lint.passed", result.lint_passed)
            root_span.set_attribute("pr.url", result.pr_url or "")
            root_span.set_attribute("task.duration_s", result.duration_s)
            if usage:
                root_span.set_attribute("agent.input_tokens", usage.input_tokens)
                root_span.set_attribute("agent.output_tokens", usage.output_tokens)
                root_span.set_attribute(
                    "agent.cache_read_input_tokens",
                    usage.cache_read_input_tokens,
                )
                root_span.set_attribute(
                    "agent.cache_creation_input_tokens",
                    usage.cache_creation_input_tokens,
                )
            if result.status != "success":
                root_span.set_status(StatusCode.ERROR, str(result.error or "task did not succeed"))

            # Emit metrics to CloudWatch Logs and print summary to stdout
            print_metrics(result_dict)

            # Persist terminal state to DynamoDB
            terminal_status = "COMPLETED" if overall_status == "success" else "FAILED"
            task_state.write_terminal(config.task_id, terminal_status, result_dict)

            return result_dict

        except Exception as e:
            # Ensure the task is marked FAILED in DynamoDB even if the pipeline
            # crashes before reaching the normal terminal-state write.
            #
            # K2 review Finding #1 — crash-path trace upload. The
            # trajectory accumulator is exactly the artifact the user
            # enabled ``--trace`` to capture the failure with; dropping
            # it on the crash path is a silent regression against the
            # design intent. Fully wrapped in its own try/except so a
            # trace upload failure cannot mask or replace the real
            # exception (we re-raise ``e`` at the end).
            crash_trace_s3_uri: str | None = None
            try:
                crash_trace_s3_uri = _maybe_upload_trace(config, trajectory, progress)
            except Exception as upload_exc:
                log(
                    "WARN",
                    f"Crash-path trace upload failed: {type(upload_exc).__name__}: {upload_exc}",
                )

            agent_for_chain = agent_result
            combined = _chain_prior_agent_error(agent_for_chain, e)
            crash_result = TaskResult(
                status="error",
                error=combined,
                task_id=config.task_id,
                agent_status=agent_for_chain.status if agent_for_chain else "unknown",
                trace_s3_uri=crash_trace_s3_uri,
            )
            task_state.write_terminal(config.task_id, "FAILED", crash_result.model_dump())
            # Best-effort ❌ on the Linear issue so the stale 👀 doesn't linger.
            # No-op for non-Linear tasks; network/GraphQL failures are swallowed.
            # `linear_eyes_reaction_id` is initialized to None at the top of
            # this try block, so it's always bound here even if we crashed
            # before the start-reaction call assigned a real id.
            react_task_finished(
                config.channel_source,
                config.channel_metadata,
                success=False,
                started_reaction_id=linear_eyes_reaction_id,
            )
            raise


def main():
    config = get_config()

    print("Task configuration loaded.", flush=True)
    print("Dry run mode detected.", flush=True)
    print()

    if config.dry_run:
        # Context hydration for dry run
        if config.issue_number:
            config.issue = fetch_github_issue(
                config.repo_url, config.issue_number, config.github_token
            )
        prompt = assemble_prompt(config)
        system_prompt = SYSTEM_PROMPT.replace("{repo_url}", config.repo_url)
        system_prompt = system_prompt.replace("{task_id}", config.task_id)
        system_prompt = system_prompt.replace("{workspace}", AGENT_WORKSPACE)
        system_prompt = system_prompt.replace("{branch_name}", "bgagent/{task_id}/dry-run")
        system_prompt = system_prompt.replace("{default_branch}", "main")
        system_prompt = system_prompt.replace("{max_turns}", str(config.max_turns))
        system_prompt = system_prompt.replace("{setup_notes}", "(dry run — setup not executed)")
        system_prompt = system_prompt.replace("{memory_context}", "(dry run — memory not loaded)")
        overrides = config.system_prompt_overrides
        if overrides:
            system_prompt += f"\n\n## Additional instructions\n\n{overrides}"
        system_prompt_hash = hashlib.sha256(system_prompt.encode("utf-8")).hexdigest()[:12]
        prompt_hash = hashlib.sha256(prompt.encode("utf-8")).hexdigest()[:12]
        print("\n--- SYSTEM PROMPT (REDACTED) ---")
        print(
            f"length={len(system_prompt)} chars sha256={system_prompt_hash} "
            "(set DEBUG_DRY_RUN_PROMPTS=1 to print full text)",
            flush=True,
        )
        print("\n--- USER PROMPT (REDACTED) ---")
        print(
            f"length={len(prompt)} chars sha256={prompt_hash} "
            "(set DEBUG_DRY_RUN_PROMPTS=1 to print full text)",
            flush=True,
        )
        if os.environ.get("DEBUG_DRY_RUN_PROMPTS") == "1":
            print(
                "\nDEBUG_DRY_RUN_PROMPTS=1 is set, but full prompt printing is disabled "
                "for secure logging compliance.",
                flush=True,
            )
        print("\n--- DRY RUN COMPLETE ---")
        return

    # Run the full pipeline.  run_task() is sync and calls asyncio.run()
    # internally, so main() must NOT be async (nested asyncio.run() is illegal).
    result = run_task(
        repo_url=config.repo_url,
        task_description=config.task_description,
        issue_number=config.issue_number,
        github_token=config.github_token,
        anthropic_model=config.anthropic_model,
        max_turns=config.max_turns,
        max_budget_usd=config.max_budget_usd,
        aws_region=config.aws_region,
        system_prompt_overrides=config.system_prompt_overrides,
        trace=config.trace,
        user_id=config.user_id,
    )

    # Exit with error if agent failed
    if result["status"] != "success":
        sys.exit(1)


if __name__ == "__main__":
    main()