feat(ecs): make the ECS Fargate compute backend usable (context-gated wiring + ECS-parity fixes)

agent/Dockerfile

@@ -44,6 +44,15 @@ RUN curl -fsSL https://deb.nodesource.com/setup_24.x | bash - && \
     apt-get clean && \
     rm -rf /var/lib/apt/lists/* /var/cache/apt/archives/*
 
+# Enable Corepack so ``yarn`` / ``pnpm`` resolve out of the box. Many target
+# repos (including ABCA itself) drive installs with ``yarn install`` via their
+# build command; without this, ``yarn`` is "command not found" (exit 127) and
+# the build-verification GATE runs inert — a green build is reported for a repo
+# we never actually built (live-caught 2026-06-29 dogfooding ABCA-on-ABCA: the
+# agent had to hand-build a ``~/bin/yarn`` shim every run). Corepack ships with
+# Node 24; ``enable`` installs the yarn/pnpm shims onto PATH.
+RUN corepack enable && corepack prepare yarn@stable --activate || corepack enable
+
 # Install Claude Code CLI (the Python SDK requires this binary)
 # Then update known vulnerable transitive packages where fixed versions exist.
 # Pinned 2.1.191 to match the CLI bundled by claude-agent-sdk 0.2.110 (see

agent/src/entrypoint.py

@@ -31,7 +31,7 @@
     TaskResult,
     TokenUsage,
 )
-from pipeline import main, run_task  # noqa: F401
+from pipeline import main, run_task, run_task_from_payload  # noqa: F401
 from post_hooks import (  # noqa: F401
     ensure_committed,
     ensure_pr,

agent/src/pipeline.py

@@ -4,6 +4,7 @@
 
 import asyncio
 import hashlib
+import inspect
 import os
 import subprocess
 import sys
@@ -1272,6 +1273,112 @@ def _on_trace_truncated(max_bytes: int, first_dropped: int) -> None:
             raise
 
 
+#: Orchestrator payload keys that map to a differently-named ``run_task`` kwarg.
+#: The orchestrator emits ``prompt``/``model_id``; ``run_task`` calls them
+#: ``task_description``/``anthropic_model``. Everything else is a 1:1 name match.
+_PAYLOAD_KEY_ALIASES = {
+    "prompt": "task_description",
+    "model_id": "anthropic_model",
+}
+
+#: ``run_task`` kwargs that must be coerced to ``str`` — the orchestrator may
+#: emit them as numbers (issue_number, pr_number) and ``run_task`` types them as
+#: strings. ``max_turns`` is coerced to int. Absent keys are left to the
+#: ``run_task`` defaults.
+_PAYLOAD_STR_KEYS = frozenset({"issue_number", "pr_number"})
+
+#: Parameter names ``run_task`` accepts — computed once at import from the REAL
+#: signature (not inside the function, so patching ``run_task`` in tests can't
+#: shadow it). Any payload key not in this set is ignored, never passed through.
+_RUN_TASK_PARAMS = frozenset(inspect.signature(run_task).parameters)
+
+#: Orchestrator payload keys we KNOW about that ``run_task`` does not (yet)
+#: accept as a parameter. Dropping one of these is expected today (e.g.
+#: ``base_branch`` is consumed via ``hydrated_context``, not a run_task kwarg;
+#: ``github_token_secret_arn`` is resolved before this call), but a key that
+#: shows up here AND is silently dropped is exactly the "wired one side of an
+#: orchestrator→agent field, forgot the other" no-op that ABCA-487 was — so we
+#: WARN when we drop one, making a future contract gap visible instead of silent.
+#: Keys not in this set (genuinely foreign) are dropped quietly as before.
+_KNOWN_ORCHESTRATOR_KEYS = frozenset(
+    {
+        "build_command",
+        # ``lint_command``'s sibling: neither is a run_task param today (the build/
+        # lint commands are consumed via repo config, not passed through here), but
+        # listing both makes a future "wired build_command, forgot lint_command"
+        # contract gap WARN instead of drop silently (N4).
+        "lint_command",
+        "merge_branches",
+        "base_branch",
+        "github_token_secret_arn",
+        # AgentCore's server.py exports task_started_at as TASK_STARTED_AT, which
+        # hooks._remaining_maxlifetime_s() uses to clip the Cedar HITL approval-gate
+        # maxLifetime. The ECS boot path bypasses server.py and does not (yet) set
+        # that env, so this key is dropped here — a silent AgentCore↔ECS HITL
+        # divergence (fail-open: the clip returns None, gate uses the task default).
+        # Listing it makes the drop WARN so the parity gap is visible until the ECS
+        # strategy sets TASK_STARTED_AT in containerEnv (tracked as a follow-up).
+        "task_started_at",
+    }
+)
+
+
+def run_task_from_payload(payload: dict) -> dict:
+    """Invoke :func:`run_task` from a full orchestrator payload dict.
+
+    The ECS compute path (``ecs-strategy.ts``) hands the agent the *entire*
+    orchestrator payload (via the #502 S3 pointer). Previously the ECS boot
+    command hand-listed a subset of ``run_task`` kwargs and silently dropped the
+    rest — most visibly ``channel_source``/``channel_metadata`` (no Linear/Jira
+    reactions or channel MCP on ECS — ABCA-487), plus ``build_command``,
+    ``cedar_policies``, ``base_branch``/``merge_branches``, ``attachments``, etc.
+
+    This maps the payload to ``run_task``'s real signature so no field can be
+    silently dropped again: rename the aliased keys, filter to parameters
+    ``run_task`` actually accepts (unknown keys are ignored, not passed as
+    ``**kwargs`` which ``run_task`` doesn't accept), and coerce the str/int
+    fields the orchestrator may emit as numbers. ``aws_region`` falls back to the
+    ``AWS_REGION`` env var when the payload omits it (the boot command used to
+    supply this explicitly).
+
+    Single source of truth + unit-testable, replacing the untestable inline
+    Python string that already drifted once.
+    """
+    kwargs: dict = {}
+    for key, value in (payload or {}).items():
+        target = _PAYLOAD_KEY_ALIASES.get(key, key)
+        if target not in _RUN_TASK_PARAMS:
+            # Not a run_task parameter — ignore. A KNOWN orchestrator key being
+            # dropped is expected today but worth a breadcrumb: if run_task ever
+            # grows a matching param, this WARN is where a "forgot to wire it
+            # through" no-op surfaces (N4 / ABCA-487 class). Foreign keys are
+            # dropped quietly.
+            if key in _KNOWN_ORCHESTRATOR_KEYS and value is not None:
+                log(
+                    "WARN",
+                    f"run_task_from_payload: dropping known orchestrator key '{key}' "
+                    f"(not a run_task parameter) — consumed elsewhere or not yet wired",
+                )
+            continue
+        if value is None:
+            continue  # let run_task's default apply
+        if target in _PAYLOAD_STR_KEYS:
+            value = str(value)
+        elif target == "max_turns":
+            # Defensive, matching how every other field is handled: a malformed
+            # max_turns must not crash the whole boot — drop it and let run_task's
+            # default apply (with a breadcrumb) rather than raise a ValueError.
+            try:
+                value = int(value)
+            except (TypeError, ValueError):
+                log("WARN", f"run_task_from_payload: ignoring non-integer max_turns {value!r}")
+                continue
+        kwargs[target] = value
+
+    kwargs.setdefault("aws_region", os.environ.get("AWS_REGION", ""))
+    return run_task(**kwargs)
+
+
 def main():
     config = get_config()
 

agent/tests/test_run_task_from_payload.py

@@ -0,0 +1,186 @@
+"""Unit tests for pipeline.run_task_from_payload — the ECS payload→run_task map.
+
+Regression cover for ABCA-487: the ECS boot command used to hand-list a subset
+of run_task kwargs and silently dropped channel_source/channel_metadata (no
+Linear/Jira reactions on ECS), build_command, cedar_policies, base_branch, etc.
+run_task_from_payload maps the WHOLE payload so nothing is dropped again.
+"""
+
+from __future__ import annotations
+
+from unittest.mock import patch
+
+from pipeline import _RUN_TASK_PARAMS, run_task_from_payload
+
+
+def _capture(payload: dict) -> dict:
+    """Run the mapper with run_task replaced by a capturing stub; return kwargs."""
+    seen: dict = {}
+
+    def fake_run_task(**kwargs):
+        seen.update(kwargs)
+        return {"status": "success"}
+
+    with patch("pipeline.run_task", side_effect=fake_run_task):
+        run_task_from_payload(payload)
+    return seen
+
+
+class TestRunTaskFromPayload:
+    def test_renames_prompt_and_model_id(self):
+        seen = _capture({"prompt": "do the thing", "model_id": "anthropic.claude-x"})
+        assert seen["task_description"] == "do the thing"
+        assert seen["anthropic_model"] == "anthropic.claude-x"
+        # The original payload keys must NOT leak through as-is (run_task rejects them).
+        assert "prompt" not in seen
+        assert "model_id" not in seen
+
+    def test_forwards_channel_fields_ABCA_487(self):
+        # THE regression: channel_source/channel_metadata must reach run_task so
+        # the Linear/Jira reaction + channel MCP fire on ECS.
+        cm = {"linear_issue_id": "iss-1", "linear_oauth_secret_arn": "arn:sm:...:lin"}
+        seen = _capture({"channel_source": "linear", "channel_metadata": cm})
+        assert seen["channel_source"] == "linear"
+        assert seen["channel_metadata"] == cm
+
+    def test_forwards_cedar_attachments_trace_user_fields(self):
+        # HITL guardrails (cedar_policies, approval_*), attachments, trace, and
+        # user_id are real run_task params here — they must reach run_task on ECS
+        # (the hand-listed boot command used to drop them).
+        seen = _capture(
+            {
+                "cedar_policies": ["p1", "p2"],
+                "attachments": [{"filename": "x.png"}],
+                "trace": True,
+                "user_id": "user-9",
+            }
+        )
+        assert seen["cedar_policies"] == ["p1", "p2"]
+        assert seen["attachments"] == [{"filename": "x.png"}]
+        assert seen["trace"] is True
+        assert seen["user_id"] == "user-9"
+
+    def test_drops_payload_keys_that_are_not_yet_run_task_params(self):
+        # build_command/lint_command (configurable verify, #1) and base_branch/
+        # merge_branches (orchestration stacking, #247) are emitted by the
+        # orchestrator but are NOT run_task parameters on this branch. The mapper
+        # filters against run_task's REAL signature, so they are dropped rather
+        # than smuggled through as an invalid kwarg. When those params land,
+        # they forward automatically with no change here — that's the point of
+        # keying off inspect.signature instead of a hand-list.
+        seen = _capture(
+            {
+                "repo_url": "org/repo",
+                "build_command": "npm ci && npm test",
+                "lint_command": "npm run lint",
+                "base_branch": "epic-tip",
+                "merge_branches": ["a", "b"],
+            }
+        )
+        assert seen["repo_url"] == "org/repo"
+        for not_yet in ("build_command", "lint_command", "base_branch", "merge_branches"):
+            assert not_yet not in seen
+
+    def test_coerces_issue_and_pr_number_to_str_and_max_turns_to_int(self):
+        seen = _capture({"issue_number": 42, "pr_number": 7, "max_turns": "50"})
+        assert seen["issue_number"] == "42"
+        assert seen["pr_number"] == "7"
+        assert seen["max_turns"] == 50
+        assert isinstance(seen["max_turns"], int)
+
+    def test_ignores_unknown_payload_keys(self):
+        # github_token_secret_arn is on the payload but is NOT a run_task param
+        # (it's consumed platform-side); passing it as **kwargs would TypeError.
+        seen = _capture(
+            {
+                "repo_url": "org/repo",
+                "github_token_secret_arn": "arn:...",
+                "sources": ["x"],
+            }
+        )
+        assert seen["repo_url"] == "org/repo"
+        assert "github_token_secret_arn" not in seen
+        assert "sources" not in seen
+
+    def test_drops_none_values_so_run_task_defaults_apply(self):
+        seen = _capture({"repo_url": "org/repo", "base_branch": None, "channel_metadata": None})
+        assert "base_branch" not in seen
+        assert "channel_metadata" not in seen
+
+    def test_aws_region_falls_back_to_env(self, monkeypatch):
+        monkeypatch.setenv("AWS_REGION", "us-east-1")
+        seen = _capture({"repo_url": "org/repo"})
+        assert seen["aws_region"] == "us-east-1"
+
+    def test_explicit_aws_region_in_payload_wins(self, monkeypatch):
+        monkeypatch.setenv("AWS_REGION", "us-east-1")
+        seen = _capture({"repo_url": "org/repo", "aws_region": "eu-west-1"})
+        assert seen["aws_region"] == "eu-west-1"
+
+    def test_every_forwarded_key_is_a_real_run_task_param(self):
+        # Guard: whatever the mapper forwards must be accepted by run_task, so a
+        # future payload key can never smuggle an invalid kwarg through. Compare
+        # against the module's real param set (run_task is patched in _capture).
+        accepted = _RUN_TASK_PARAMS
+        seen = _capture(
+            {
+                "prompt": "p",
+                "model_id": "m",
+                "repo_url": "r",
+                "issue_number": 1,
+                "channel_source": "linear",
+                "channel_metadata": {"a": "b"},
+                "build_command": "b",
+                "cedar_policies": ["c"],
+                "base_branch": "x",
+                "attachments": [{}],
+                "trace": False,
+                "user_id": "u",
+                "pr_number": 3,
+                "hydrated_context": {"k": "v"},
+                "resolved_workflow": {"id": "w"},
+            }
+        )
+        assert set(seen).issubset(accepted)
+
+    def test_warns_when_dropping_a_known_orchestrator_key(self):
+        # N4: a KNOWN orchestrator key that run_task doesn't accept is dropped
+        # (expected today) but logged, so a future "wired one side, forgot the
+        # other" contract gap (the ABCA-487 class) is visible, not silent.
+        logs: list[tuple[str, str]] = []
+        with patch("pipeline.log", side_effect=lambda level, msg, **kw: logs.append((level, msg))):
+            _capture({"build_command": "mise run build", "repo_url": "r"})
+        assert [m for level, m in logs if level == "WARN" and "build_command" in m], (
+            "expected a WARN when dropping the known orchestrator key build_command"
+        )
+
+    def test_does_NOT_warn_when_dropping_a_foreign_key(self):
+        # A genuinely-foreign key (not a known orchestrator field) is dropped
+        # quietly — no log noise for keys we never expected to forward.
+        logs: list[tuple[str, str]] = []
+        with patch("pipeline.log", side_effect=lambda level, msg, **kw: logs.append((level, msg))):
+            _capture({"some_future_unrelated_key": "v", "repo_url": "r"})
+        assert not [m for level, m in logs if "some_future_unrelated_key" in m]
+
+    def test_warns_when_dropping_task_started_at_HITL_parity(self):
+        # task_started_at drives the AgentCore HITL maxLifetime clip via
+        # TASK_STARTED_AT; the ECS path doesn't set it yet, so its drop must WARN
+        # (surface the AgentCore↔ECS parity gap, not silently fail-open).
+        logs: list[tuple[str, str]] = []
+        with patch("pipeline.log", side_effect=lambda level, msg, **kw: logs.append((level, msg))):
+            _capture({"task_started_at": "2026-07-14T00:00:00Z", "repo_url": "r"})
+        assert [m for level, m in logs if level == "WARN" and "task_started_at" in m]
+
+    def test_malformed_max_turns_is_dropped_not_raised(self):
+        # A non-integer max_turns must not crash the boot — it's dropped (run_task
+        # default applies) with a WARN, matching how every other field defaults.
+        logs: list[tuple[str, str]] = []
+        with patch("pipeline.log", side_effect=lambda level, msg, **kw: logs.append((level, msg))):
+            seen = _capture({"repo_url": "r", "max_turns": "not-a-number"})
+        assert "max_turns" not in seen
+        assert [m for level, m in logs if level == "WARN" and "max_turns" in m]
+
+    def test_valid_max_turns_still_coerced_to_int(self):
+        seen = _capture({"repo_url": "r", "max_turns": "50"})
+        assert seen["max_turns"] == 50
+        assert isinstance(seen["max_turns"], int)

cdk/mise.toml

@@ -40,8 +40,17 @@ run = ["mkdir -p $TMPDIR", "yarn synth:quiet"]
 description = "cdk deploy (pass args after --)"
 run = "npx cdk deploy"
 
+# Bootstraps with ComputeTypes=agentcore (the template default). To ALSO enable the
+# ECS compute backend you must set the ComputeTypes CFN *parameter* — `cdk bootstrap`
+# has no way to pass template parameters (no --parameters flag; --context sets CDK
+# construct context, NOT a CFN parameter), so run this task once, then set the
+# parameter directly on the CDKToolkit stack:
+#   aws cloudformation update-stack --stack-name CDKToolkit --use-previous-template \
+#     --capabilities CAPABILITY_NAMED_IAM \
+#     --parameters ParameterKey=ComputeTypes,ParameterValue=agentcore\,ecs
+#   # verify: aws cloudformation describe-stacks --stack-name CDKToolkit --query 'Stacks[0].Parameters'
 [tasks.bootstrap]
-description = "Bootstrap CDK with least-privilege policies (pass -- --context ComputeTypes=agentcore,ecs for ECS)"
+description = "Bootstrap CDK with least-privilege policies (ComputeTypes=agentcore; for ECS set the ComputeTypes CFN param on CDKToolkit — see comment above the task)"
 depends = [":install", ":bootstrap:generate"]
 run = "npx cdk bootstrap --template bootstrap/bootstrap-template.yaml"