feat(harness): Phase A — headless run entrypoint + Run now (PR-1)

.github/workflows/platform.yml

@@ -96,6 +96,12 @@ jobs:
       # `CDK_KB_SYNC_ENABLED` variable to "false" in an environment only to
       # dark-stop the feature there.
       CDK_KB_SYNC_ENABLED: ${{ vars.CDK_KB_SYNC_ENABLED }}
+      # Scheduled runs (headless "Run now" + grant routes, and the Phase-B
+      # scheduler when it lands). Default ON with a kill switch: unset
+      # resolves to empty string, which config.ts treats as the default
+      # (on). Set the `CDK_SCHEDULED_RUNS_ENABLED` variable to "false" in
+      # an environment only to dark-stop the feature there.
+      CDK_SCHEDULED_RUNS_ENABLED: ${{ vars.CDK_SCHEDULED_RUNS_ENABLED }}
       # Secrets
       AWS_ROLE_ARN: ${{ secrets.AWS_ROLE_ARN }}
       AWS_ACCESS_KEY_ID: ${{ secrets.AWS_ACCESS_KEY_ID }}

backend/scripts/spike_headless_run.py

@@ -0,0 +1,249 @@
+"""Dev-ai driver for the headless run-entrypoint spike (F1).
+
+Runs `apis.shared.harness.run_agent_headless` from a laptop against the
+deployed dev-ai AgentCore Runtime — through the runtime gateway, exactly the
+path a scheduler worker would take. See
+docs/specs/harness-entrypoint-spike-findings.md.
+
+Usage (requires an authenticated AWS profile for the dev-ai account):
+
+    cd backend
+    AWS_PROFILE=dev-ai uv run python scripts/spike_headless_run.py \
+        --user-id <cognito-sub> \
+        --prompt "Find 3-credit undergraduate communication classes" \
+        --tools class_search
+
+    # Negative probes for the record (gateway auth evidence):
+    AWS_PROFILE=dev-ai uv run python scripts/spike_headless_run.py \
+        --user-id <cognito-sub> --probe-workload-token --probe-sigv4
+
+The script resolves all names from SSM / naming conventions for --prefix
+(default dev-boisestateai-v2), exports the env vars the shared harness
+expects, runs the turn, then reads back the session-metadata row and the
+RUN# audit record as delivery proof.
+"""
+
+from __future__ import annotations
+
+import argparse
+import asyncio
+import json
+import logging
+import os
+import sys
+import urllib.parse
+import uuid
+
+logging.basicConfig(level=logging.INFO, format="%(levelname)s %(name)s: %(message)s")
+logger = logging.getLogger("spike")
+
+
+def resolve_environment(prefix: str, region: str) -> dict:
+    """Resolve dev-ai names from SSM + conventions and export harness env."""
+    import boto3
+
+    ssm = boto3.client("ssm", region_name=region)
+    sts = boto3.client("sts", region_name=region)
+    account = sts.get_caller_identity()["Account"]
+
+    runtime_id = ssm.get_parameter(Name=f"/{prefix}/inference-api/runtime-id")[
+        "Parameter"
+    ]["Value"]
+    runtime_arn = f"arn:aws:bedrock-agentcore:{region}:{account}:runtime/{runtime_id}"
+    client_id = ssm.get_parameter(Name=f"/{prefix}/auth/cognito/bff-app-client-id")[
+        "Parameter"
+    ]["Value"]
+
+    env = {
+        "AWS_REGION": region,
+        "INFERENCE_API_URL": (
+            f"https://bedrock-agentcore.{region}.amazonaws.com/runtimes/{runtime_arn}"
+        ),
+        "BFF_SESSIONS_TABLE_NAME": f"{prefix}-bff-sessions",
+        "COGNITO_BFF_APP_CLIENT_ID": client_id,
+        "COGNITO_BFF_APP_CLIENT_SECRET_ARN": f"{prefix}-cognito-bff-app-client-secret",
+        "DYNAMODB_SESSIONS_METADATA_TABLE_NAME": f"{prefix}-sessions-metadata",
+    }
+    os.environ.update(env)
+    return {"runtime_arn": runtime_arn, "account": account, **env}
+
+
+def probe_workload_token(prefix: str, region: str, runtime_arn: str, user_id: str) -> None:
+    """Unknown-1 'try first' path — recorded evidence: the gateway rejects it."""
+    import boto3
+    import httpx
+
+    client = boto3.client("bedrock-agentcore", region_name=region)
+    token = client.get_workload_access_token_for_user_id(
+        workloadName=f"{prefix}-platform-workload", userId=user_id
+    )["workloadAccessToken"]
+    is_jwt = token.count(".") == 2
+    logger.info("workload token minted (len=%d, jwt=%s)", len(token), is_jwt)
+
+    encoded = urllib.parse.quote(runtime_arn, safe="")
+    url = (
+        f"https://bedrock-agentcore.{region}.amazonaws.com/runtimes/"
+        f"{encoded}/invocations?qualifier=DEFAULT"
+    )
+    r = httpx.post(
+        url,
+        headers={"Authorization": f"Bearer {token}"},
+        json={"session_id": f"probe-{uuid.uuid4().hex[:8]}", "message": "ping"},
+        timeout=30,
+    )
+    logger.info("PROBE workload-token bearer -> HTTP %d %s", r.status_code, r.text[:200])
+
+
+def probe_sigv4(region: str, runtime_arn: str) -> None:
+    """IAM data-plane call — recorded evidence: authorizer-method mismatch."""
+    import boto3
+
+    client = boto3.client("bedrock-agentcore", region_name=region)
+    try:
+        resp = client.invoke_agent_runtime(
+            agentRuntimeArn=runtime_arn,
+            qualifier="DEFAULT",
+            runtimeSessionId=f"probe-sigv4-{uuid.uuid4().hex}",
+            contentType="application/json",
+            accept="text/event-stream",
+            payload=json.dumps(
+                {"session_id": f"probe-{uuid.uuid4().hex[:8]}", "message": "ping"}
+            ).encode(),
+        )
+        logger.info("PROBE sigv4 -> statusCode=%s", resp.get("statusCode"))
+    except Exception as exc:
+        logger.info("PROBE sigv4 -> %s: %s", type(exc).__name__, exc)
+
+
+def verify_delivery(prefix: str, region: str, user_id: str, session_id: str, run_id: str) -> None:
+    """Read back the session row + audit record as F2/F6a proof."""
+    import boto3
+    from boto3.dynamodb.conditions import Key
+
+    table = boto3.resource("dynamodb", region_name=region).Table(
+        f"{prefix}-sessions-metadata"
+    )
+    rows = table.query(
+        IndexName="SessionLookupIndex",
+        KeyConditionExpression=Key("GSI_PK").eq(f"SESSION#{session_id}"),
+    )["Items"]
+    meta = [r for r in rows if str(r.get("SK", "")).startswith("S#")]
+    messages = [r for r in rows if str(r.get("GSI_SK", "")).startswith("C#")]
+    logger.info(
+        "DELIVERY session row: %s",
+        json.dumps(
+            {
+                k: str(v)
+                for k, v in (meta[0] if meta else {}).items()
+                if k in ("title", "status", "messageCount", "lastModel", "SK")
+            }
+        ),
+    )
+    logger.info("DELIVERY persisted message items: %d", len(messages))
+
+    audit = table.get_item(
+        Key={"PK": f"USER#{user_id}", "SK": f"RUN#{run_id}"}
+    ).get("Item")
+    logger.info(
+        "AUDIT record: %s",
+        json.dumps({k: str(v) for k, v in (audit or {}).items()}, sort_keys=True)[:600],
+    )
+
+
+async def main() -> int:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--prefix", default="dev-boisestateai-v2")
+    parser.add_argument("--region", default="us-west-2")
+    parser.add_argument("--user-id", required=True, help="Cognito sub of the run owner")
+    parser.add_argument("--prompt", default="Reply with the single word: pong")
+    parser.add_argument(
+        "--tools",
+        default=None,
+        help="Comma-separated enabled_tools (omit for the user's defaults)",
+    )
+    parser.add_argument("--title", default=None, help="Explicit session title")
+    parser.add_argument("--probe-workload-token", action="store_true")
+    parser.add_argument("--probe-sigv4", action="store_true")
+    parser.add_argument("--skip-run", action="store_true")
+    args = parser.parse_args()
+
+    resolved = resolve_environment(args.prefix, args.region)
+    logger.info("runtime: %s", resolved["runtime_arn"])
+
+    if args.probe_workload_token:
+        probe_workload_token(
+            args.prefix, args.region, resolved["runtime_arn"], args.user_id
+        )
+    if args.probe_sigv4:
+        probe_sigv4(args.region, resolved["runtime_arn"])
+    if args.skip_run:
+        return 0
+
+    # Import after env export — the harness reads configuration from env.
+    from apis.shared.harness import (
+        CognitoRefreshBearerAuth,
+        HeadlessGrantService,
+        run_agent_headless,
+    )
+
+    # Dev-driver stand-in for create-on-enable: production creates the
+    # headless grant from the caller's *live* session on the "Run now"
+    # route; from a laptop we bootstrap it from the user's newest BFF
+    # session row instead (a filtered Scan is fine for a dev script).
+    grants = HeadlessGrantService()
+    if await grants.get_active_grant(args.user_id) is None:
+        import boto3
+        from boto3.dynamodb.conditions import Attr
+
+        table = boto3.resource("dynamodb", region_name=args.region).Table(
+            os.environ["BFF_SESSIONS_TABLE_NAME"]
+        )
+        rows: list[dict] = []
+        kwargs: dict = {"FilterExpression": Attr("user_id").eq(args.user_id)}
+        while True:
+            page = table.scan(**kwargs)
+            rows.extend(page.get("Items", []))
+            if "LastEvaluatedKey" not in page:
+                break
+            kwargs["ExclusiveStartKey"] = page["LastEvaluatedKey"]
+        if not rows:
+            logger.error(
+                "No BFF session for %s — log in once, then re-run", args.user_id
+            )
+            return 1
+        newest = max(rows, key=lambda r: int(r.get("last_seen_at") or 0))
+        await grants.enable(
+            user_id=args.user_id,
+            username=str(newest["username"]),
+            refresh_token=str(newest["cognito_refresh_token"]),
+            token_issued_at=int(newest.get("created_at") or 0) or None,
+        )
+        logger.info("Bootstrapped headless grant for %s", args.user_id)
+
+    async def on_event(name: str, data: dict) -> None:
+        if name in ("tool_use", "tool_result", "session_title", "stream_error"):
+            logger.info("SSE %s: %s", name, json.dumps(data, default=str)[:220])
+
+    result = await run_agent_headless(
+        user_id=args.user_id,
+        prompt=args.prompt,
+        auth=CognitoRefreshBearerAuth(),
+        enabled_tools=args.tools.split(",") if args.tools else None,
+        agent_type="chat",
+        trigger="spike",
+        title=args.title,
+        on_event=on_event,
+    )
+
+    print("\n================ RunResult ================")
+    print(json.dumps(result.to_dict(), indent=2, default=str)[:4000])
+    print("===========================================\n")
+
+    verify_delivery(
+        args.prefix, args.region, args.user_id, result.session_id, result.run_id
+    )
+    return 0 if result.status == "completed" else 1
+
+
+if __name__ == "__main__":
+    sys.exit(asyncio.run(main()))

backend/src/apis/app_api/chat/proxy_routes.py

@@ -21,10 +21,9 @@
 import httpx
 from fastapi import APIRouter, Depends, HTTPException, Request
 from fastapi.responses import StreamingResponse
-from urllib.parse import quote, urlsplit
-
 from apis.shared.auth.dependencies import get_current_user_from_session
 from apis.shared.auth.models import User
+from apis.shared.harness.runner import build_invocations_url
 
 logger = logging.getLogger(__name__)
 
@@ -37,29 +36,12 @@ def _inference_api_url() -> str:
 # wedged upstream eventually surfaces.
 _PROXY_TIMEOUT_SECONDS = 300.0
 
-
-def _build_invocations_url(base_url: str) -> str:
-    """Resolve the upstream `/invocations` URL from `INFERENCE_API_URL`.
-
-    Cloud: `INFERENCE_API_URL` is the AgentCore Runtime data-plane base
-    (`https://bedrock-agentcore.<region>.amazonaws.com/runtimes/<ARN>`),
-    where `<ARN>` is unencoded in SSM. The data-plane route is
-    `POST /runtimes/{agentRuntimeArn}/invocations?qualifier={qualifier}`
-    with `{agentRuntimeArn}` as a single URL-encoded path segment — so the
-    ARN's literal `/` and `:` must be percent-encoded or AWS returns 404.
-    A `qualifier` is also required; we use `DEFAULT`.
-
-    Local dev: `INFERENCE_API_URL` is `http://localhost:8001`, where
-    `/invocations` is a real FastAPI route on inference-api directly. No
-    encoding or qualifier needed.
-    """
-    parts = urlsplit(base_url)
-    prefix = "/runtimes/"
-    if parts.netloc.startswith("bedrock-agentcore.") and parts.path.startswith(prefix):
-        arn = parts.path[len(prefix):]
-        encoded_arn = quote(arn, safe="")
-        return f"{parts.scheme}://{parts.netloc}/runtimes/{encoded_arn}/invocations?qualifier=DEFAULT"
-    return f"{base_url}/invocations"
+# Canonical `/invocations` URL resolution lives in the shared harness
+# (`apis.shared.harness.runner.build_invocations_url`) — the headless
+# runner, this proxy, and the MCP Apps proxy all share one copy. Kept
+# under the historical private name so existing call sites and docstring
+# references stay valid.
+_build_invocations_url = build_invocations_url
 
 
 def _build_upstream_client() -> httpx.AsyncClient:

backend/src/apis/app_api/main.py

@@ -200,6 +200,7 @@ async def lifespan(app: FastAPI):
 from apis.app_api.voice import router as voice_router
 from apis.app_api.user_menu_links.routes import router as user_menu_links_router
 from apis.app_api.system_prompts.routes import router as system_prompts_router
+from apis.app_api.runs.routes import router as runs_router
 
 # Include routers
 app.include_router(health_router)
@@ -233,6 +234,7 @@ async def lifespan(app: FastAPI):
 app.include_router(voice_router)  # Cookie-authenticated WS proxy for Nova Sonic voice mode (#211)
 app.include_router(user_menu_links_router)  # Public read of admin-managed user-menu links
 app.include_router(system_prompts_router)   # Public read of admin-managed system prompts
+app.include_router(runs_router)  # Headless "Run now" + grant lifecycle (scheduled-runs PR-1; SCHEDULED_RUNS_ENABLED + RBAC gated at runtime)
 
 # Conditionally register fine-tuning routes
 if os.environ.get("FINE_TUNING_ENABLED", "false").lower() == "true":

backend/src/apis/app_api/runs/__init__.py

@@ -0,0 +1 @@
+"""User-facing headless-run surface ("Run now" + headless-grant lifecycle)."""