From 9111a8378f2e32893a623a7b990dc1f076f0646a Mon Sep 17 00:00:00 2001 From: Phil Merrell Date: Sun, 5 Jul 2026 15:36:57 -0600 Subject: [PATCH 1/4] =?UTF-8?q?feat(harness):=20spike=20headless=20agent-r?= =?UTF-8?q?un=20entrypoint=20(F1)=20=E2=80=94=20proven=20in=20dev-ai?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit run_agent_headless in apis/shared/harness: per-owner Cognito bearer mint (workload-token + SigV4 front-door paths proven dead at the runtime gateway), server-side SSE drain pinned to live wire shapes, F6a audit records + guardrails/classification seams, delivery via the runtime's own session materialization + title override. Driver script reproduces the dev-ai proof and the negative auth probes. Findings + Phase A design in docs/specs/harness-entrypoint-spike-findings.md. Co-Authored-By: Claude Opus 4.8 --- backend/scripts/spike_headless_run.py | 211 ++++++++++++++++ backend/src/apis/shared/harness/__init__.py | 44 ++++ backend/src/apis/shared/harness/auth.py | 151 ++++++++++++ backend/src/apis/shared/harness/governance.py | 179 ++++++++++++++ backend/src/apis/shared/harness/models.py | 66 +++++ backend/src/apis/shared/harness/runner.py | 225 +++++++++++++++++ backend/src/apis/shared/harness/sse.py | 227 ++++++++++++++++++ backend/tests/apis/shared/test_harness_sse.py | 157 ++++++++++++ .../harness-entrypoint-spike-findings.md | 131 ++++++++++ 9 files changed, 1391 insertions(+) create mode 100644 backend/scripts/spike_headless_run.py create mode 100644 backend/src/apis/shared/harness/__init__.py create mode 100644 backend/src/apis/shared/harness/auth.py create mode 100644 backend/src/apis/shared/harness/governance.py create mode 100644 backend/src/apis/shared/harness/models.py create mode 100644 backend/src/apis/shared/harness/runner.py create mode 100644 backend/src/apis/shared/harness/sse.py create mode 100644 backend/tests/apis/shared/test_harness_sse.py create mode 100644 docs/specs/harness-entrypoint-spike-findings.md diff --git a/backend/scripts/spike_headless_run.py b/backend/scripts/spike_headless_run.py new file mode 100644 index 000000000..124335501 --- /dev/null +++ b/backend/scripts/spike_headless_run.py @@ -0,0 +1,211 @@ +"""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 \ + --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 --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, run_agent_headless + + 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())) diff --git a/backend/src/apis/shared/harness/__init__.py b/backend/src/apis/shared/harness/__init__.py new file mode 100644 index 000000000..2b26bdc19 --- /dev/null +++ b/backend/src/apis/shared/harness/__init__.py @@ -0,0 +1,44 @@ +"""Headless agent-run harness (F1 spike). + +The trigger-agnostic entrypoint for running an agent turn as a user with no +live browser session: schedules, "Run now", A2A, webhooks, and eval harnesses +are all just callers of :func:`run_agent_headless`. + +Lives in ``apis.shared`` because it is consumed by more than one service +(app-api "Run now" routes, future dispatcher/worker Lambdas) and must never +be an inference-api route — the AgentCore Runtime data plane only exposes +``/invocations`` + ``/ping`` (see CLAUDE.md, Inference API boundary). The +harness is a *client* of ``/invocations``, not a new server surface. + +Spike status: see docs/specs/harness-entrypoint-spike-findings.md. +""" + +from apis.shared.harness.auth import ( + BearerAuthStrategy, + CognitoRefreshBearerAuth, + HeadlessAuthError, + StaticBearerAuth, +) +from apis.shared.harness.governance import GovernanceFloor, RunAuditRecorder +from apis.shared.harness.models import ( + OAuthConsentRequired, + RunResult, + RunStatus, + ToolTraceEntry, +) +from apis.shared.harness.runner import build_invocations_url, run_agent_headless + +__all__ = [ + "BearerAuthStrategy", + "CognitoRefreshBearerAuth", + "GovernanceFloor", + "HeadlessAuthError", + "OAuthConsentRequired", + "RunAuditRecorder", + "RunResult", + "RunStatus", + "StaticBearerAuth", + "ToolTraceEntry", + "build_invocations_url", + "run_agent_headless", +] diff --git a/backend/src/apis/shared/harness/auth.py b/backend/src/apis/shared/harness/auth.py new file mode 100644 index 000000000..b7ff4a030 --- /dev/null +++ b/backend/src/apis/shared/harness/auth.py @@ -0,0 +1,151 @@ +"""Unattended bearer minting for headless runs (Unknown 1). + +The AgentCore Runtime is provisioned with a **Cognito customJwtAuthorizer** +(`inference-agentcore-construct.ts`: discovery URL = the user pool, +`allowedClients` = [BFF app client]). Spike probes against dev-ai proved: + +- A platform **workload access token** (`GetWorkloadAccessTokenForUserId`) + is NOT accepted as the `/invocations` bearer — it is an opaque encrypted + blob, not a JWT; the gateway rejects it with + ``403 {"message": "OAuth authorization failed: Failed to parse token"}``. +- **SigV4** (`invoke_agent_runtime`) is also rejected once a JWT authorizer + is configured: ``AccessDeniedException: Authorization method mismatch``. + +So the only front door is a real Cognito access token for the owning user, +minted by the platform. :class:`CognitoRefreshBearerAuth` implements the +zero-infra-change path: exchange the user's stored BFF refresh token +(`REFRESH_TOKEN_AUTH` + SECRET_HASH — the exact machinery +`SessionRefreshMiddleware` already uses) for a fresh 1-hour access token. +The pool does not rotate refresh tokens (CDK default), so the mint does not +disturb the user's live browser sessions. + +The workload identity is still essential — but one layer down: *inside* the +runtime, connector tokens are minted from the vault via +`GetWorkloadAccessTokenForUserId` keyed by the `sub` of the bearer we send +(`apis/shared/oauth/agentcore_identity.py`). The front-door bearer and the +vault leg are two different trust boundaries; the spike brief's "try first" +path conflated them. +""" + +from __future__ import annotations + +import asyncio +import logging +import os +from typing import Optional, Protocol + +import boto3 + +from apis.shared.sessions_bff.refresh import CognitoRefreshClient, CognitoRefreshError + +logger = logging.getLogger(__name__) + + +class HeadlessAuthError(RuntimeError): + """No bearer could be minted for the requested user. + + For a scheduled trigger this should pause the schedule (analogous to + KB-sync's ``paused_reauth``) — the user must log in again to renew the + grant. + """ + + +class BearerAuthStrategy(Protocol): + """Seam between the runner and however a bearer is obtained. + + Phase A can add strategies (e.g. a dedicated headless-grant record, or + an M2M client + trusted `user_id` payload) without touching the runner. + """ + + async def mint_bearer_for_user(self, user_id: str) -> str: ... + + +class StaticBearerAuth: + """Wrap an already-obtained token (tests; callers with a live token).""" + + def __init__(self, token: str) -> None: + self._token = token + + async def mint_bearer_for_user(self, user_id: str) -> str: + return self._token + + +class CognitoRefreshBearerAuth: + """Mint a per-owner access token from the user's stored refresh token. + + Reads the newest BFF session row for ``user_id`` and runs the standard + Cognito refresh exchange. Requires the caller's IAM principal to read + the BFF sessions table and the app-client secret — app-api's role + already holds both grants. + + Spike-scoped caveats (Phase A must address): + - The BFF sessions table is keyed by ``session_id`` only; finding a + user's row is a filtered ``Scan``. Fine at spike scale; Phase A needs + a ``user_id`` GSI or (better) a dedicated headless-grant record that + stores a refresh token minted at opt-in time with its own lifecycle. + - The grant inherits BFF session lifetime (~30-day absolute cap + + row TTL): a user who hasn't logged in recently cannot be run + headlessly. That is arguably the *right* governance default, but it + must be an explicit product decision, not an accident. + """ + + def __init__( + self, + *, + sessions_table_name: Optional[str] = None, + refresh_client: Optional[CognitoRefreshClient] = None, + region: Optional[str] = None, + ) -> None: + self._table_name = sessions_table_name or os.environ.get( + "BFF_SESSIONS_TABLE_NAME", "" + ) + self._refresh_client = refresh_client or CognitoRefreshClient() + self._region = region or os.environ.get("AWS_REGION", "us-west-2") + + def _newest_session_row(self, user_id: str) -> Optional[dict]: + if not self._table_name: + raise HeadlessAuthError("BFF_SESSIONS_TABLE_NAME is not configured") + table = boto3.resource("dynamodb", region_name=self._region).Table( + self._table_name + ) + from boto3.dynamodb.conditions import Attr + + rows: list[dict] = [] + kwargs = {"FilterExpression": Attr("user_id").eq(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: + return None + return max(rows, key=lambda r: int(r.get("last_seen_at") or 0)) + + async def mint_bearer_for_user(self, user_id: str) -> str: + row = await asyncio.to_thread(self._newest_session_row, user_id) + if row is None: + raise HeadlessAuthError( + f"No stored BFF session for user {user_id}; the user must " + "log in before headless runs can act as them." + ) + try: + refreshed = await self._refresh_client.refresh( + username=str(row["username"]), + refresh_token=str(row["cognito_refresh_token"]), + ) + except CognitoRefreshError as exc: + raise HeadlessAuthError( + f"Cognito refused the refresh exchange for user {user_id}: {exc}" + ) from exc + if refreshed.refresh_token != str(row["cognito_refresh_token"]): + # Rotation is off on this pool; if it is ever enabled this mint + # would invalidate the user's browser session unless persisted. + logger.warning( + "Cognito rotated the refresh token during a headless mint for " + "user %s — the stored BFF session row is now stale. Enable " + "rotation-aware persistence before using this strategy on a " + "rotating pool.", + user_id, + ) + return refreshed.access_token diff --git a/backend/src/apis/shared/harness/governance.py b/backend/src/apis/shared/harness/governance.py new file mode 100644 index 000000000..13cb04f42 --- /dev/null +++ b/backend/src/apis/shared/harness/governance.py @@ -0,0 +1,179 @@ +"""Governance floor (F6a) seams for headless runs. + +The moment an agent turn runs unattended *as a user*, it touches user data +with no human in the loop — so every headless run passes through this floor. +The spike implements exactly one slice for real (the audit record); the +guardrails and data-classification checkpoints are explicit no-op seams so +Phase A fills them in without touching the runner's control flow. + +Hook points (called by ``run_agent_headless`` in this order): + +1. ``on_run_start`` — AUDIT (implemented): who/what/when/trigger, before + any token is minted or model called. +2. ``check_input`` — GUARDRAILS seam (no-op): Phase A calls Bedrock + ``ApplyGuardrail`` on the prompt; a blocked verdict raises before the + run spends tokens or reads user data. +3. ``classify_output`` — DATA-CLASSIFICATION seam (no-op): Phase A runs the + PII/FERPA checkpoint over the final message (and tool-result previews) + before the result is delivered anywhere. +4. ``on_run_end`` — AUDIT (implemented): outcome, stop reason, tool + names, usage. + +Audit records are written to the sessions-metadata table under +``PK=USER#{user_id}, SK=RUN#{run_id}``. The session listing queries +``begins_with(SK, 'S#ACTIVE#')`` and the GSI lookups use their own key +shapes, so ``RUN#`` items are invisible to every existing access path. +Phase A should promote run-records to their own table (or a documented SK +family) with a "due/recent runs" GSI; for the spike the point is that the +seam exists and every run leaves a durable, queryable trail. +""" + +from __future__ import annotations + +import hashlib +import logging +import os +from datetime import datetime, timezone +from typing import Any, Dict, Optional + +import boto3 + +from apis.shared.harness.models import RunResult + +logger = logging.getLogger(__name__) + + +def _now_iso() -> str: + return datetime.now(timezone.utc).isoformat().replace("+00:00", "Z") + + +class RunAuditRecorder: + """Minimal durable audit trail for headless runs (the implemented slice).""" + + def __init__( + self, *, table_name: Optional[str] = None, region: Optional[str] = None + ) -> None: + self._table_name = table_name or os.environ.get( + "DYNAMODB_SESSIONS_METADATA_TABLE_NAME", "" + ) + self._region = region or os.environ.get("AWS_REGION", "us-west-2") + + def _table(self): + if not self._table_name: + raise RuntimeError( + "DYNAMODB_SESSIONS_METADATA_TABLE_NAME is required for run auditing" + ) + return boto3.resource("dynamodb", region_name=self._region).Table( + self._table_name + ) + + def record_start( + self, + *, + run_id: str, + user_id: str, + session_id: str, + trigger: str, + prompt: str, + ) -> None: + self._table().put_item( + Item={ + "PK": f"USER#{user_id}", + "SK": f"RUN#{run_id}", + "runId": run_id, + "userId": user_id, + "sessionId": session_id, + "trigger": trigger, + "promptSha256": hashlib.sha256(prompt.encode()).hexdigest(), + "promptChars": len(prompt), + "status": "started", + "startedAt": _now_iso(), + } + ) + + def record_end(self, *, result: RunResult) -> None: + self._table().update_item( + Key={"PK": f"USER#{result.user_id}", "SK": f"RUN#{result.run_id}"}, + UpdateExpression=( + "SET #s = :s, stopReason = :sr, errorDetail = :e, " + "toolNames = :t, #u = :u, finishedAt = :f, " + "finalMessageChars = :c" + ), + ExpressionAttributeNames={"#s": "status", "#u": "usage"}, + ExpressionAttributeValues={ + ":s": result.status, + ":sr": result.stop_reason or "", + ":e": (result.error or "")[:1000], + ":t": [t.name for t in result.tool_trace], + # Dynamo rejects floats; usage payloads are ints but guard + # against provider metrics like latency ratios. + ":u": _dynamo_safe(result.usage), + ":f": _now_iso(), + ":c": len(result.final_message), + }, + ) + + +def _dynamo_safe(value: Any) -> Any: + if isinstance(value, float): + return str(value) + if isinstance(value, dict): + return {k: _dynamo_safe(v) for k, v in value.items()} + if isinstance(value, list): + return [_dynamo_safe(v) for v in value] + return value + + +class GovernanceFloor: + """F6a checkpoint bundle every headless run passes through.""" + + def __init__(self, *, audit: Optional[RunAuditRecorder] = None) -> None: + self._audit = audit or RunAuditRecorder() + + async def on_run_start( + self, + *, + run_id: str, + user_id: str, + session_id: str, + trigger: str, + prompt: str, + ) -> None: + """AUDIT — implemented. Failure here fails the run (a run that can't + be audited must not execute unattended).""" + self._audit.record_start( + run_id=run_id, + user_id=user_id, + session_id=session_id, + trigger=trigger, + prompt=prompt, + ) + + async def check_input(self, *, prompt: str, user_id: str) -> None: + """GUARDRAILS seam — no-op in the spike. + + Phase A: Bedrock ``ApplyGuardrail`` (source=INPUT) on the prompt; + raise a ``GovernanceBlocked`` error on a blocked verdict so the + runner records an audited, non-retryable failure. + """ + + async def classify_output(self, *, result: RunResult) -> None: + """DATA-CLASSIFICATION seam — no-op in the spike. + + Phase A: PII/FERPA checkpoint over ``result.final_message`` and + ``result.tool_trace`` previews before delivery. May redact (mutate + the result) or block (raise), per policy. + """ + + async def on_run_end(self, *, result: RunResult) -> None: + """AUDIT — implemented. Best-effort: a failed end-write must not + destroy an otherwise-delivered result (the start record already + pins the run's existence).""" + try: + self._audit.record_end(result=result) + except Exception: + logger.error( + "Failed to write run-end audit record for %s", + result.run_id, + exc_info=True, + ) diff --git a/backend/src/apis/shared/harness/models.py b/backend/src/apis/shared/harness/models.py new file mode 100644 index 000000000..3e1882b1c --- /dev/null +++ b/backend/src/apis/shared/harness/models.py @@ -0,0 +1,66 @@ +"""Result shapes for headless agent runs. + +``RunResult`` is the structured return every trigger consumes (schedule +worker, "Run now" route, future A2A server front). Keep it JSON-friendly: +``asdict(result)`` must serialize cleanly so it can become a run-record +item or an A2A task artifact without translation. +""" + +from __future__ import annotations + +from dataclasses import asdict, dataclass, field +from typing import Any, Dict, List, Literal, Optional + +# completed — stream drained to `done` with no stream-level error +# error — HTTP error, `stream_error`/`error` event, or transport failure +# timeout — the SSE stream exceeded the caller's budget +# oauth_required — the turn finished but at least one connector tool needs +# user consent (headless runs cannot pop a consent window; +# callers should surface the authorization URL to the user) +RunStatus = Literal["completed", "error", "timeout", "oauth_required"] + + +@dataclass +class ToolTraceEntry: + """One tool invocation observed on the stream.""" + + tool_use_id: str + name: str + input: Dict[str, Any] = field(default_factory=dict) + result_preview: Optional[str] = None + is_error: bool = False + + +@dataclass +class OAuthConsentRequired: + """An `oauth_required` SSE event — a connector needs (re-)consent.""" + + provider_id: str + authorization_url: str + + +@dataclass +class RunResult: + """Structured outcome of one headless agent turn.""" + + run_id: str + session_id: str + user_id: str + status: RunStatus + final_message: str = "" + stop_reason: Optional[str] = None + error: Optional[str] = None + title: Optional[str] = None + tool_trace: List[ToolTraceEntry] = field(default_factory=list) + # Accumulated usage/metrics from the stream's `metadata_summary` (turn + # totals) with per-message `metadata` events as fallback. Shape mirrors + # the SSE payloads: {"usage": {...}, "metrics": {...}}. + usage: Dict[str, Any] = field(default_factory=dict) + oauth_required: List[OAuthConsentRequired] = field(default_factory=list) + started_at: str = "" + finished_at: str = "" + # Diagnostic: counts of every SSE event name seen, e.g. {"tool_use": 2}. + events_seen: Dict[str, int] = field(default_factory=dict) + + def to_dict(self) -> Dict[str, Any]: + return asdict(self) diff --git a/backend/src/apis/shared/harness/runner.py b/backend/src/apis/shared/harness/runner.py new file mode 100644 index 000000000..19412d10e --- /dev/null +++ b/backend/src/apis/shared/harness/runner.py @@ -0,0 +1,225 @@ +"""`run_agent_headless` — the F1 headless run entrypoint (spike). + +Given ``(user_id, prompt)`` and no live session: mint a per-owner bearer, +POST the AgentCore Runtime ``/invocations`` data plane, drain the SSE stream +server-side, pass the governance floor, and land the result as a retrievable +session. Every trigger (schedule worker, "Run now" route, A2A server, +webhook) is just a caller of this function. + +A2A-readiness: the seam is ``(user_id, prompt, resolved config) → RunResult`` +plus the optional ``on_event`` callback, which relays every typed SSE event +as it arrives — an A2A server front can map that stream onto task status +updates +without a rewrite. Reminder from CLAUDE.md: if this is ever exposed as an +A2A server, its advertised ``capabilities`` MUST include ``streaming=True``. +""" + +from __future__ import annotations + +import logging +import os +import uuid +from datetime import datetime, timezone +from typing import Any, Awaitable, Callable, Dict, List, Optional +from urllib.parse import quote, urlsplit + +import httpx + +from apis.shared.harness.auth import BearerAuthStrategy, HeadlessAuthError +from apis.shared.harness.governance import GovernanceFloor +from apis.shared.harness.models import RunResult +from apis.shared.harness.sse import InvocationStreamAccumulator, iter_sse_events + +logger = logging.getLogger(__name__) + +# Matches the app-api chat proxy's budget for a full agent turn; the runtime +# data plane itself enforces a hard cap in the same range. +DEFAULT_TIMEOUT_SECONDS = 300.0 + +OnEvent = Callable[[str, Dict[str, Any]], Awaitable[None]] + + +def _now_iso() -> str: + return datetime.now(timezone.utc).isoformat().replace("+00:00", "Z") + + +def build_invocations_url(base_url: str) -> str: + """Resolve the upstream `/invocations` URL from an INFERENCE_API_URL-style base. + + Same resolution as the app-api chat proxy (`proxy_routes.py`): in cloud + the base is the AgentCore data plane + (`https://bedrock-agentcore..amazonaws.com/runtimes/`) whose + ARN path segment must be percent-encoded and given a qualifier; locally + it's a plain FastAPI origin. Phase A should point the proxy at this + shared copy rather than keeping two. + """ + 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" + + +async def run_agent_headless( + *, + user_id: str, + prompt: str, + auth: BearerAuthStrategy, + session_id: Optional[str] = None, + run_id: Optional[str] = None, + title: Optional[str] = None, + model_id: Optional[str] = None, + rag_assistant_id: Optional[str] = None, + enabled_tools: Optional[List[str]] = None, + agent_type: Optional[str] = None, + inference_params: Optional[Dict[str, Any]] = None, + trigger: str = "manual", + invocations_base_url: Optional[str] = None, + timeout_seconds: float = DEFAULT_TIMEOUT_SECONDS, + governance: Optional[GovernanceFloor] = None, + on_event: Optional[OnEvent] = None, +) -> RunResult: + """Run one agent turn as ``user_id`` with no live session. + + The run-config parameters (``model_id``, ``rag_assistant_id``, + ``enabled_tools``, ``agent_type``, ``inference_params``) mirror the + existing ``InvocationRequest`` contract — the entrypoint resolves *no* + new config type (scheduled-agent-runs.md decision #6). + + Returns a ``RunResult`` in all outcomes except audit-start failure and + ``HeadlessAuthError`` (both raise: a run we cannot audit or authenticate + must not execute). + """ + run_id = run_id or f"run-{uuid.uuid4().hex[:12]}" + session_id = session_id or f"headless-{uuid.uuid4().hex[:16]}" + governance = governance or GovernanceFloor() + started_at = _now_iso() + + # F6a checkpoints 1 + 2 — before any token mint or model spend. + await governance.on_run_start( + run_id=run_id, + user_id=user_id, + session_id=session_id, + trigger=trigger, + prompt=prompt, + ) + await governance.check_input(prompt=prompt, user_id=user_id) + + result = RunResult( + run_id=run_id, + session_id=session_id, + user_id=user_id, + status="error", + started_at=started_at, + ) + + try: + bearer = await auth.mint_bearer_for_user(user_id) + except HeadlessAuthError: + result.finished_at = _now_iso() + result.error = "auth: could not mint a bearer for the user" + await governance.on_run_end(result=result) + raise + + base_url = invocations_base_url or os.environ.get( + "INFERENCE_API_URL", "http://localhost:8001" + ) + url = build_invocations_url(base_url) + payload: Dict[str, Any] = { + "session_id": session_id, + "message": prompt, + } + if model_id is not None: + payload["model_id"] = model_id + if rag_assistant_id is not None: + payload["rag_assistant_id"] = rag_assistant_id + if enabled_tools is not None: + payload["enabled_tools"] = enabled_tools + if agent_type is not None: + payload["agent_type"] = agent_type + if inference_params is not None: + payload["inference_params"] = inference_params + + acc = InvocationStreamAccumulator() + try: + async with httpx.AsyncClient( + timeout=httpx.Timeout(timeout_seconds) + ) as client: + async with client.stream( + "POST", + url, + headers={ + "Content-Type": "application/json", + "Authorization": f"Bearer {bearer}", + }, + json=payload, + ) as response: + if response.status_code >= 400: + body = await response.aread() + result.error = ( + f"HTTP {response.status_code}: " + f"{body.decode('utf-8', errors='replace')[:500]}" + ) + else: + async for name, data in iter_sse_events( + response.aiter_lines() + ): + acc.handle(name, data) + if on_event is not None: + await on_event(name, data) + except httpx.TimeoutException: + result.status = "timeout" + result.error = f"stream exceeded {timeout_seconds:.0f}s budget" + except httpx.HTTPError as exc: + result.error = f"transport: {type(exc).__name__}: {exc}" + + result.final_message = acc.final_message + result.stop_reason = acc.stop_reason + result.tool_trace = acc.tool_trace + result.usage = acc.finalize_usage() + result.oauth_required = acc.oauth_required + result.title = acc.title + result.events_seen = acc.events_seen + result.error = result.error or acc.error + if result.status != "timeout": + if result.error: + result.status = "error" + elif acc.oauth_required: + result.status = "oauth_required" + elif acc.done: + result.status = "completed" + else: + result.status = "error" + result.error = "stream ended without a done event" + result.finished_at = _now_iso() + + # F6a checkpoint 3 — classify before delivery. + await governance.classify_output(result=result) + + # Delivery (minimal F2): the runtime already persisted the session row, + # messages, and an auto-generated title during the turn. Make the row's + # existence unconditional (idempotent belt-and-braces for error paths) + # and let an explicit caller title win over the generated one. + try: + from apis.shared.sessions.metadata import ( + ensure_session_metadata_exists, + update_session_title, + ) + + await ensure_session_metadata_exists(session_id, user_id) + if title: + await update_session_title(session_id, user_id, title) + result.title = title + except Exception: + logger.error( + "Headless run %s: result delivery failed", run_id, exc_info=True + ) + + # F6a checkpoint 4 — outcome audit (best-effort inside). + await governance.on_run_end(result=result) + return result diff --git a/backend/src/apis/shared/harness/sse.py b/backend/src/apis/shared/harness/sse.py new file mode 100644 index 000000000..cc4781936 --- /dev/null +++ b/backend/src/apis/shared/harness/sse.py @@ -0,0 +1,227 @@ +"""Server-side SSE consumption for `/invocations` streams (Unknown 2). + +Nothing else in the platform reads an `/invocations` SSE stream server-side — +the app-api chat proxy only relays bytes to a browser. This module parses the +stream into `(event_name, payload)` pairs and accumulates them into the +fields a `RunResult` needs. + +Wire format (see `agents/main_agent/streaming/stream_processor.py` and the +SSE table in CLAUDE.md): `event: \\ndata: \\n\\n`. A few legacy +paths emit bare `data:` lines whose JSON carries a `type` field — the parser +falls back to that. The stream interleaves raw Strands passthrough events +(`event: event`) with the typed events; the accumulator only consumes the +typed ones. +""" + +from __future__ import annotations + +import json +import logging +from dataclasses import dataclass, field +from typing import Any, AsyncIterator, Dict, List, Optional, Tuple + +from apis.shared.harness.models import OAuthConsentRequired, ToolTraceEntry + +logger = logging.getLogger(__name__) + +_TOOL_RESULT_PREVIEW_CHARS = 2000 + + +async def iter_sse_events( + lines: AsyncIterator[str], +) -> AsyncIterator[Tuple[str, Dict[str, Any]]]: + """Parse an SSE line stream into `(event_name, payload)` tuples. + + Follows the SSE framing rules we actually emit: `event:` (optional) and + `data:` lines terminated by a blank line. Multi-`data:`-line events are + joined per the SSE spec. Events with unparseable JSON payloads are + surfaced as `("_unparseable", {"raw": ...})` so callers can count them + without the reader dying mid-stream. + """ + event_name: Optional[str] = None + data_lines: List[str] = [] + + async for raw_line in lines: + line = raw_line.rstrip("\n") + if line == "": + if data_lines or event_name is not None: + data = "\n".join(data_lines) + payload: Dict[str, Any] + try: + payload = json.loads(data) if data else {} + if not isinstance(payload, dict): + payload = {"value": payload} + except json.JSONDecodeError: + yield "_unparseable", {"raw": data[:500]} + event_name, data_lines = None, [] + continue + # Bare `data:` events carry their name in a `type` field. + name = event_name or str(payload.get("type") or "message") + yield name, payload + event_name, data_lines = None, [] + continue + if line.startswith("event:"): + event_name = line[len("event:"):].strip() + elif line.startswith("data:"): + data_lines.append(line[len("data:"):].lstrip()) + # Comments (`:`) and unknown fields are ignored per the SSE spec. + + +@dataclass +class InvocationStreamAccumulator: + """Folds the typed event stream into RunResult-shaped state. + + Text accumulation: assistant text arrives as `content_block_delta` + events (`type == "text"`); a turn with tool use emits several + `message_start`/`message_stop` cycles, so the *final* assistant message + is the text of the last completed message (falling back to any + unterminated buffer, then to the whole-turn transcript). + """ + + done: bool = False + stop_reason: Optional[str] = None + error: Optional[str] = None + title: Optional[str] = None + tool_trace: List[ToolTraceEntry] = field(default_factory=list) + oauth_required: List[OAuthConsentRequired] = field(default_factory=list) + usage: Dict[str, Any] = field(default_factory=dict) + events_seen: Dict[str, int] = field(default_factory=dict) + + _current: List[str] = field(default_factory=list) + _messages: List[str] = field(default_factory=list) + _trace_by_id: Dict[str, ToolTraceEntry] = field(default_factory=dict) + _per_message_usage: Dict[str, Any] = field(default_factory=dict) + + @property + def final_message(self) -> str: + for text in reversed(self._messages + ["".join(self._current)]): + if text.strip(): + return text + return "" + + @property + def transcript(self) -> str: + parts = [m for m in self._messages if m.strip()] + tail = "".join(self._current) + if tail.strip(): + parts.append(tail) + return "\n\n".join(parts) + + def handle(self, name: str, payload: Dict[str, Any]) -> None: + self.events_seen[name] = self.events_seen.get(name, 0) + 1 + + if name == "message_start": + if "".join(self._current).strip(): + self._messages.append("".join(self._current)) + self._current = [] + elif name == "content_block_delta": + if payload.get("type") == "text" and payload.get("text"): + self._current.append(str(payload["text"])) + elif name == "message_stop": + self.stop_reason = payload.get("stopReason") or self.stop_reason + if "".join(self._current).strip(): + self._messages.append("".join(self._current)) + self._current = [] + elif name == "tool_use": + # Two wire shapes: the flat event-formatter payload + # ({toolUseId, name, input}) and the stream-processor passthrough + # ({"tool_use": {tool_use_id, name, input}}) where `input` is a + # *partial JSON string* re-emitted as the model streams the + # arguments. Upsert by id so a streamed tool call folds into one + # trace entry whose input is the last parseable prefix. + data = payload.get("tool_use") + if not isinstance(data, dict): + data = payload + tool_use_id = str( + data.get("toolUseId") or data.get("tool_use_id") or "" + ) + entry = self._trace_by_id.get(tool_use_id) + if entry is None: + entry = ToolTraceEntry(tool_use_id=tool_use_id, name="") + self.tool_trace.append(entry) + if tool_use_id: + self._trace_by_id[tool_use_id] = entry + if data.get("name"): + entry.name = str(data["name"]) + raw_input = data.get("input") + if isinstance(raw_input, dict): + entry.input = raw_input + elif isinstance(raw_input, str) and raw_input: + try: + parsed = json.loads(raw_input) + if isinstance(parsed, dict): + entry.input = parsed + except json.JSONDecodeError: + pass # partial prefix; a later re-emit will complete it + elif name in ("tool_result", "tool_error"): + # Flat event-formatter payload ({toolUseId, result}) or the + # message-shaped passthrough ({"message": {"content": + # [{"toolResult": {toolUseId, status, content: [{text}]}}]}}). + tool_results: List[Dict[str, Any]] = [] + message = payload.get("message") + if isinstance(message, dict): + for block in message.get("content") or []: + if isinstance(block, dict) and isinstance( + block.get("toolResult"), dict + ): + tool_results.append(block["toolResult"]) + if not tool_results: + tool_results.append(payload) + for tr in tool_results: + tool_use_id = str( + tr.get("toolUseId") or tr.get("tool_use_id") or "" + ) + entry = self._trace_by_id.get(tool_use_id) + if entry is None and self.tool_trace: + entry = self.tool_trace[-1] + if entry is None: + continue + result = tr.get("result") or tr.get("error") + if result is None and isinstance(tr.get("content"), list): + result = "\n".join( + str(block.get("text")) + for block in tr["content"] + if isinstance(block, dict) and "text" in block + ) + entry.result_preview = str(result or "")[ + :_TOOL_RESULT_PREVIEW_CHARS + ] + if name == "tool_error" or tr.get("status") == "error": + entry.is_error = True + elif name == "metadata": + # Per-model-call usage; keep the last as a fallback if the turn + # summary never arrives (short turns emit both). + for key in ("usage", "metrics"): + if key in payload: + self._per_message_usage[key] = payload[key] + elif name == "metadata_summary": + # Turn-cumulative totals — authoritative for cost attribution. + self.usage.update( + {k: v for k, v in payload.items() if k != "type"} + ) + elif name == "session_title": + self.title = payload.get("title") or self.title + elif name == "oauth_required": + provider = str( + payload.get("providerId") or payload.get("provider_id") or "" + ) + url = str( + payload.get("authorizationUrl") + or payload.get("authorization_url") + or "" + ) + if url: + self.oauth_required.append( + OAuthConsentRequired(provider_id=provider, authorization_url=url) + ) + elif name in ("stream_error", "error"): + self.error = str( + payload.get("message") or payload.get("error") or payload + )[:2000] + elif name == "done": + self.done = True + + def finalize_usage(self) -> Dict[str, Any]: + if self.usage: + return self.usage + return dict(self._per_message_usage) diff --git a/backend/tests/apis/shared/test_harness_sse.py b/backend/tests/apis/shared/test_harness_sse.py new file mode 100644 index 000000000..88168b92c --- /dev/null +++ b/backend/tests/apis/shared/test_harness_sse.py @@ -0,0 +1,157 @@ +"""Tests for the harness SSE reader/accumulator (headless run spike). + +Event payload shapes below are verbatim captures from a live dev-ai +`/invocations` stream (2026-07-05, spike run `run-8f10d164cff9`), so the +accumulator is pinned to what the runtime actually emits — including the +stream-processor's nested `tool_use` passthrough whose `input` is a +partial JSON *string* re-emitted as the model streams arguments, and the +message-shaped `tool_result`. +""" + +import pytest + +from apis.shared.harness.sse import InvocationStreamAccumulator, iter_sse_events + + +async def _aiter(lines): + for line in lines: + yield line + + +@pytest.mark.asyncio +async def test_iter_sse_events_parses_named_and_bare_data_events(): + lines = [ + "event: message_start", + 'data: {"role": "assistant"}', + "", + # Bare data event (event_formatter path) — name comes from `type`. + 'data: {"type": "session_title", "title": "T"}', + "", + "event: done", + "data: {}", + "", + ] + events = [pair async for pair in iter_sse_events(_aiter(lines))] + assert events == [ + ("message_start", {"role": "assistant"}), + ("session_title", {"type": "session_title", "title": "T"}), + ("done", {}), + ] + + +@pytest.mark.asyncio +async def test_iter_sse_events_survives_unparseable_payload(): + lines = ["event: event", "data: {not json", "", "event: done", "data: {}", ""] + events = [pair async for pair in iter_sse_events(_aiter(lines))] + assert events[0][0] == "_unparseable" + assert events[-1] == ("done", {}) + + +def _drive_turn_events(): + """A tool-use turn in the shapes the runtime actually emits.""" + return [ + ("message_start", {"role": "assistant"}), + # Streamed partial tool input: same id re-emitted with growing input. + ( + "tool_use", + {"tool_use": {"name": "search_classes", "tool_use_id": "t1", "input": ""}}, + ), + ( + "tool_use", + { + "tool_use": { + "name": "search_classes", + "tool_use_id": "t1", + "input": '{"subject": "CO', + } + }, + ), + ( + "tool_use", + { + "tool_use": { + "name": "search_classes", + "tool_use_id": "t1", + "input": '{"subject": "COMM", "min_credits": 3}', + } + }, + ), + ("message_stop", {"stopReason": "tool_use"}), + ( + "tool_result", + { + "message": { + "role": "user", + "content": [ + { + "toolResult": { + "status": "success", + "toolUseId": "t1", + "content": [{"text": '{"total_results": 87}'}], + } + } + ], + } + }, + ), + ("message_start", {"role": "assistant"}), + ("content_block_delta", {"contentBlockIndex": 0, "type": "text", "text": "Found "}), + ("content_block_delta", {"contentBlockIndex": 0, "type": "text", "text": "87 classes."}), + ("message_stop", {"stopReason": "end_turn"}), + ( + "metadata", + {"usage": {"inputTokens": 5, "outputTokens": 215, "totalTokens": 10763}}, + ), + ("session_title", {"type": "session_title", "title": "COMM Search"}), + ("done", {}), + ] + + +def test_accumulator_folds_streamed_tool_use_into_one_entry(): + acc = InvocationStreamAccumulator() + for name, payload in _drive_turn_events(): + acc.handle(name, payload) + + assert acc.done is True + assert acc.stop_reason == "end_turn" + assert acc.final_message == "Found 87 classes." + assert acc.title == "COMM Search" + + assert len(acc.tool_trace) == 1 + entry = acc.tool_trace[0] + assert entry.tool_use_id == "t1" + assert entry.name == "search_classes" + assert entry.input == {"subject": "COMM", "min_credits": 3} + assert entry.result_preview == '{"total_results": 87}' + assert entry.is_error is False + + usage = acc.finalize_usage() + assert usage["usage"]["totalTokens"] == 10763 + + +def test_accumulator_flat_tool_shapes_and_errors(): + acc = InvocationStreamAccumulator() + acc.handle("tool_use", {"toolUseId": "t2", "name": "calc", "input": {"x": 1}}) + acc.handle("tool_result", {"toolUseId": "t2", "result": "boom", "status": "error"}) + acc.handle("stream_error", {"message": "model exploded"}) + + assert acc.tool_trace[0].input == {"x": 1} + assert acc.tool_trace[0].is_error is True + assert acc.tool_trace[0].result_preview == "boom" + assert acc.error == "model exploded" + assert acc.done is False + + +def test_accumulator_oauth_required_and_final_message_fallback(): + acc = InvocationStreamAccumulator() + acc.handle("message_start", {"role": "assistant"}) + acc.handle( + "content_block_delta", {"contentBlockIndex": 0, "type": "text", "text": "partial"} + ) + acc.handle( + "oauth_required", + {"providerId": "google-drive", "authorizationUrl": "https://consent"}, + ) + # No message_stop — unterminated buffer must still surface. + assert acc.final_message == "partial" + assert acc.oauth_required[0].provider_id == "google-drive" diff --git a/docs/specs/harness-entrypoint-spike-findings.md b/docs/specs/harness-entrypoint-spike-findings.md new file mode 100644 index 000000000..0613b6639 --- /dev/null +++ b/docs/specs/harness-entrypoint-spike-findings.md @@ -0,0 +1,131 @@ +# Spike Findings — Headless Agent-Run Entrypoint (F1) + +**Status:** Spike complete — success criterion met in dev-ai +**Author:** Fable 5 (spike executed 2026-07-05, dev-ai acct 490617140655, us-west-2) +**Brief:** `docs/specs/harness-entrypoint-spike-brief.md` · **Parents:** `docs/specs/agentic-platform-primitives.md` (F1/F2/F6a), `docs/specs/scheduled-agent-runs.md` §4.2 +**Spike branch:** `spike/harness-headless-entrypoint` + +--- + +## Verdict + +**GO for Phase A.** Chosen auth path: **platform-minted per-owner Cognito access token** (the brief's "fallback"), because the preferred path is *proven impossible* against the runtime gateway as deployed — see Unknown 1. Both hard unknowns are resolved with working code, and the success criterion ran end-to-end in dev-ai: + +> Headless caller + `(user_id, prompt)`, no live session → agent turn ran **as the user** through the real AgentCore Runtime gateway → the turn called `search_classes` (the user's per-user-authorized class-search MCP connector, which validated the minted token downstream) → the **governance floor** wrote start/end audit records → the result landed as a **retrievable session** in the user's conversation list (title, metadata row, 2 persisted message items). + +Evidence run: `run-8f10d164cff9` / session `headless-09a30f6ac87b4092` / user `18419330-70a1-7018-f8e3-9577b2e18455` (2026-07-05T21:32Z). Reproduce with: + +```bash +cd backend && uv sync --extra agentcore --extra dev +AWS_PROFILE=dev-ai uv run python scripts/spike_headless_run.py \ + --user-id \ + --prompt "Use the class search tool to find two 3-credit undergraduate COMM classes..." \ + --tools class_search --title "Headless Spike — Class Search" +``` + +One scope note: the criterion's example connector ("lists a Google Drive file") could not run *inside* the turn because dev-ai has **no cloud-reachable vault-3LO agent tool** — the only 3LO tool (`canvas_faculty`) points at `http://localhost:8026`. That is a catalog gap, not an auth gap. The vault leg was proven separately and unattended: the same headless caller minted the user's **google-drive** token from the AgentCore vault via the platform workload identity (`GetWorkloadAccessTokenForUserId` → `get_token_for_user`, mirroring the consent flow's `customParameters` per the vault-key gotcha) and listed the user's actual Drive files. Connector auth is therefore proven on **both** protocols we have: forward-auth (in-run) and vault-3LO (out-of-run, identical mechanism to the in-run tool path). + +--- + +## Unknown 1 — Unattended auth to the runtime (RESOLVED: fallback path) + +The brief's decided approach — invoke `/invocations` with the **platform workload token** — **cannot work**, and not for a fixable configuration reason: + +| Probe (dev-ai, real gateway) | Result | +|---|---| +| Workload access token as `Authorization: Bearer` | **403** `{"message":"OAuth authorization failed: Failed to parse token"}`. The token from `GetWorkloadAccessTokenForUserId` is an **opaque encrypted blob** (`AgV4…`, 1864 chars, not 3-segment JWT). The gateway's JWT authorizer can't even parse it, let alone validate issuer/client. | +| SigV4 `invoke_agent_runtime` (IAM data plane) | **AccessDeniedException:** *"Authorization method mismatch. The agent is configured for a different authorization method…"* — once a `customJWTAuthorizer` is configured (ours: Cognito discovery URL, `allowedClients=[BFF app client]`, `inference-agentcore-construct.ts` ~274), IAM invocation is refused outright. | + +The brief's "try first" path conflated **two different trust boundaries**: the workload identity governs the *token vault* (connector OAuth, inside the run) — it was never a front-door credential. The front door only accepts a Cognito JWT for the allowed client. So the platform must mint a **real Cognito access token for the owning user**: + +- **Spike implementation (works today, zero infra change):** exchange the user's stored BFF refresh token (`bff-sessions` table) via `REFRESH_TOKEN_AUTH` + SECRET_HASH — the exact machinery `SessionRefreshMiddleware`/`CognitoRefreshClient` already run. Verified: mints a 1-hour token with the right `sub`; the pool does **not** rotate refresh tokens (CDK default), so the user's live browser sessions are untouched (`rotated_refresh=False` observed). +- The minted token then works **three layers deep**: gateway JWT authorizer → container `get_current_user_trusted` (`sub` → user_id, so memory/RBAC/quota/session all resolve to the right user) → forwarded to forward-auth MCP servers, which validate `client_id == BFF client` (class-search accepted it). +- The **workload identity keeps its real job** unchanged: inside the run, connector tokens mint from the vault keyed by the `sub` of our bearer — no code change needed there. + +**Phase A hardening (recommended, not blocking):** the refresh-token grant inherits BFF session lifetime (30-day absolute cap, sliding TTL) and is discovered by a table **Scan** (no user_id GSI). Ship Phase A on this path but behind an explicit **headless-grant record**: when a user enables scheduled runs, store a purpose-minted refresh token (or pin a session row) with its own lifecycle + revocation, and add a `user_id` GSI. "You must have logged in within 30 days for the platform to act as you" is a defensible governance default — make it a documented product decision. A dedicated M2M app client + trusted `user_id` payload was considered and rejected for Phase A: it requires container-side confused-deputy handling (`sub` = client-id, trust the payload) plus CDK changes to `allowedClients`, and it *weakens* the story that every downstream check sees a genuine user token. + +## Unknown 2 — Server-side SSE consumption (RESOLVED: built and pinned to live shapes) + +`apis/shared/harness/sse.py` — an httpx-based reader (`iter_sse_events`) + `InvocationStreamAccumulator` that drains to `done`, yielding a `RunResult` with final message, tool trace, usage, title, `stream_error`, and `oauth_required`. Non-obvious wire facts discovered live (unit tests pin them): + +- The stream interleaves **typed events** with raw Strands passthrough (`event: event`); consume only the typed ones. +- `tool_use` arrives as `{"tool_use": {"tool_use_id", "name", "input"}}` where `input` is a **partial JSON string re-emitted repeatedly** as the model streams arguments — fold by id, keep the last parseable prefix. `tool_result` arrives **message-shaped** (`{"message": {"content": [{"toolResult": …}]}}`), not flat. (The flat `event_formatter` shapes also exist on other paths; both are handled.) +- Turn totals come on `metadata_summary` (cumulative), not the per-call `metadata` events — use the summary for cost attribution, per-call as fallback. +- A tool-use turn emits multiple `message_start/stop` cycles; "final message" = last completed non-empty assistant text. +- `session_title` may arrive mid-stream (or after `done`) and never carries the placeholder. +- 300s budget matches the proxy; `httpx.TimeoutException` → `status="timeout"`. + +--- + +## Design deliverables + +### 1. `run_agent_headless(...)` + `RunResult` + +```python +async def run_agent_headless( + *, user_id: str, prompt: str, auth: BearerAuthStrategy, + session_id: str | None = None, run_id: str | None = None, title: str | None = None, + # run-config mirrors InvocationRequest — no new config type (decision #6): + model_id: str | None = None, rag_assistant_id: str | None = None, + enabled_tools: list[str] | None = None, agent_type: str | None = None, + inference_params: dict | None = None, + trigger: str = "manual", # audit dimension: schedule|run_now|a2a|… + invocations_base_url: str | None = None, # env INFERENCE_API_URL fallback + timeout_seconds: float = 300.0, + governance: GovernanceFloor | None = None, + on_event: OnEvent | None = None, # A2A streaming seam +) -> RunResult +``` + +`RunResult` (`apis/shared/harness/models.py`): `run_id, session_id, user_id, status ∈ {completed, error, timeout, oauth_required}, final_message, stop_reason, error, title, tool_trace[{tool_use_id, name, input, result_preview, is_error}], usage{usage,metrics}, oauth_required[{provider_id, authorization_url}], started_at, finished_at, events_seen` — `to_dict()` is JSON-clean so it can become a run-record item or A2A task artifact untranslated. + +**A2A-readiness:** the seam is `(user, input, config) → structured result` **plus** `on_event`, which relays every typed SSE event live — an A2A server front maps that onto task status updates without rewriting the runner. ⚠️ Standing CLAUDE.md rule: if exposed as A2A, `capabilities` must include `streaming=True`. `oauth_required` is a first-class status because a headless run cannot pop a consent window — schedulers should pause (KB-sync `paused_reauth` analog) and surface the URL. + +### 2. Where it lives + +**`backend/src/apis/shared/harness/`** — a shared invocation *client*, not a route. Justification against the service-boundary rules: + +- It cannot be an inference-api route (only `/invocations` + `/ping` reachable in cloud) — and it doesn't need to be: the harness **calls** `/invocations`; the agent loop is unchanged. +- Its consumers span services: app-api ("Run now" route, PR-1 of scheduled-runs), the Phase-B dispatcher/worker Lambdas, a future A2A front. "Needed by more than one → `apis.shared`" (CLAUDE.md). `tests/architecture/test_import_boundaries.py` passes. +- The dedicated-Lambda option is a *deployment* choice for Phase B (the worker imports this module), not a module-boundary choice — same code either way. +- One dedupe noted in-code: `build_invocations_url` is a copy of the proxy's resolver; Phase A should point `proxy_routes.py` at the shared copy. + +### 3. Auth approach chosen + +**Platform-minted per-owner Cognito access token** (`CognitoRefreshBearerAuth`, `apis/shared/harness/auth.py`), behind a `BearerAuthStrategy` protocol so Phase A can swap in the headless-grant record without touching the runner. Evidence and the two dead ends are in Unknown 1 above; the negative probes are reproducible via `--probe-workload-token --probe-sigv4` on the driver script. + +### 4. Governance floor (F6a) hook points + +`apis/shared/harness/governance.py` — every headless run passes four checkpoints, placed so Phase A fills them without touching runner control flow: + +1. **`on_run_start` — AUDIT (implemented).** Durable record *before* any token mint or model spend: `PK=USER#{user}, SK=RUN#{run_id}` in the sessions-metadata table — `trigger`, `promptSha256`, `promptChars`, `startedAt`, `status=started`. Invisible to all existing access paths (session listing queries `begins_with(SK,'S#ACTIVE#')`). Fail-closed: a run that can't be audited doesn't execute. +2. **`check_input` — guardrails seam (no-op).** Phase A: Bedrock `ApplyGuardrail(source=INPUT)`; blocked verdict raises before spend, leaving an audited failure. +3. **`classify_output` — data-classification seam (no-op).** Phase A: PII/FERPA checkpoint over `final_message` + tool-result previews, **before delivery**; may redact (mutate) or block (raise). +4. **`on_run_end` — AUDIT (implemented).** Outcome, stop reason, tool names, usage, finishedAt. Best-effort (a failed end-write can't destroy a delivered result; the start record pins existence). + +Phase A promotions: run-records to their own table (or documented SK family) with a "recent runs per user" GSI; add the caller's IAM identity to the start record; wire `trigger` from real callers. + +### 5. Delivery (minimal F2) + +Confirmed — and cheaper than the spec assumed: **the runtime turn itself already materializes almost everything** (`/invocations` pre-creates the session row via `ensure_session_metadata_exists`, persists user+assistant messages, generates + persists a Nova title, updates activity/costs). Verified post-run: session row `S#ACTIVE#…#headless-09a30f6ac87b4092` (status active, messageCount 1→, title), **2 message items**, session visible in the user's list. The harness adds: idempotent `ensure_session_metadata_exists` (belt-and-braces for error paths) + `update_session_title` when the caller passes an explicit title (e.g. *"Morning Briefing — Jul 6"*), which wins over the generated one. + +**Run-record shape** (the audit item; Phase A may split audit vs. delivery records): `runId, userId, sessionId, trigger, promptSha256, promptChars, status, stopReason, errorDetail, toolNames[], usage{...}, finalMessageChars, startedAt, finishedAt`. + +--- + +## Phase A punch list (from spike scars) + +1. Headless-grant record + `user_id` GSI (replace the BFF-table Scan); decide the "must have logged in within N days" policy explicitly. +2. Fill `check_input` / `classify_output`; promote run-records out of the sessions table. +3. "Run now" app-api route (cookie auth + RBAC capability) calling `run_agent_headless` — the PR-1 validation surface. +4. Dedupe `build_invocations_url` with the chat proxy. +5. Rotation-aware persistence in `CognitoRefreshBearerAuth` before anyone enables refresh-token rotation on the pool (currently warn-only). +6. Catalog gap, separate from Phase A: no cloud-reachable vault-3LO agent tool exists in dev-ai — deploy one (e.g. the canvas MCP server, or a Drive tool) so scheduled-run dogfooding exercises the 3LO path in-run. +7. `enabled_tools=None` means "all RBAC-allowed" — fine for "Run now", but schedules should snapshot an explicit tool set at creation (least surprise when the catalog changes under a sleeping schedule). + +## Files (spike branch) + +- `backend/src/apis/shared/harness/{__init__,models,sse,auth,governance,runner}.py` — the F1 primitive +- `backend/scripts/spike_headless_run.py` — dev-ai driver (positive proof + recorded negative probes) +- `backend/tests/apis/shared/test_harness_sse.py` — SSE shapes pinned from the live stream +- No new packages (httpx/boto3 already pinned); no spec files modified; no inference-api changes. From 0cd8000618bd77180c8d9ede11b6a51dbe237f70 Mon Sep 17 00:00:00 2001 From: Phil Merrell Date: Sun, 5 Jul 2026 18:10:59 -0600 Subject: [PATCH 2/4] feat(harness): headless-grant record + production hardening of the F1 primitive MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Replace the spike's BFF-table Scan with an explicit headless-grant record (apis/shared/harness/grants.py): create-on-enable from an attended session, per-owner lookup via the sparse HeadlessGrantUserIndex GSI, revocation that deletes the stored credential, and a documented "must have logged in within 30 days" policy (TTL anchored to the login that issued the pinned refresh token, matching the Cognito refresh-token validity). CognitoRefreshBearerAuth now mints from the grant (rotation-aware: a rotated refresh token is persisted back before the mint returns — punch-list #5). Dedupe build_invocations_url into the harness as the single canonical resolver; the chat proxy imports it (punch-list #4). Document the enabled_tools=None = all-RBAC-allowed semantic and the schedule-snapshot rule (punch-list #7). Governance docstrings updated to the locked F6a decision: audit-only fail-closed + wired no-op guardrail/classification seams, implementations gated to the scheduled phase. Add an _build_http_client seam and tests covering the grant lifecycle, grant-backed minting, audit fail-closed ordering, and stream outcomes against a MockTransport runtime. Co-Authored-By: Claude Opus 4.8 --- backend/scripts/spike_headless_run.py | 40 ++- backend/src/apis/app_api/chat/proxy_routes.py | 32 +- backend/src/apis/shared/harness/__init__.py | 20 +- backend/src/apis/shared/harness/auth.py | 139 ++++---- backend/src/apis/shared/harness/governance.py | 51 +-- backend/src/apis/shared/harness/grants.py | 331 ++++++++++++++++++ backend/src/apis/shared/harness/runner.py | 50 ++- .../tests/apis/shared/test_harness_grants.py | 319 +++++++++++++++++ .../tests/apis/shared/test_harness_runner.py | 272 ++++++++++++++ 9 files changed, 1112 insertions(+), 142 deletions(-) create mode 100644 backend/src/apis/shared/harness/grants.py create mode 100644 backend/tests/apis/shared/test_harness_grants.py create mode 100644 backend/tests/apis/shared/test_harness_runner.py diff --git a/backend/scripts/spike_headless_run.py b/backend/scripts/spike_headless_run.py index 124335501..a639ef254 100644 --- a/backend/scripts/spike_headless_run.py +++ b/backend/scripts/spike_headless_run.py @@ -180,7 +180,45 @@ async def main() -> int: return 0 # Import after env export — the harness reads configuration from env. - from apis.shared.harness import CognitoRefreshBearerAuth, run_agent_headless + 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"): diff --git a/backend/src/apis/app_api/chat/proxy_routes.py b/backend/src/apis/app_api/chat/proxy_routes.py index b01d036c2..9fa060d7f 100644 --- a/backend/src/apis/app_api/chat/proxy_routes.py +++ b/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..amazonaws.com/runtimes/`), - where `` 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: diff --git a/backend/src/apis/shared/harness/__init__.py b/backend/src/apis/shared/harness/__init__.py index 2b26bdc19..1070d1d10 100644 --- a/backend/src/apis/shared/harness/__init__.py +++ b/backend/src/apis/shared/harness/__init__.py @@ -1,16 +1,16 @@ -"""Headless agent-run harness (F1 spike). +"""Headless agent-run harness (primitive F1). The trigger-agnostic entrypoint for running an agent turn as a user with no live browser session: schedules, "Run now", A2A, webhooks, and eval harnesses are all just callers of :func:`run_agent_headless`. Lives in ``apis.shared`` because it is consumed by more than one service -(app-api "Run now" routes, future dispatcher/worker Lambdas) and must never -be an inference-api route — the AgentCore Runtime data plane only exposes -``/invocations`` + ``/ping`` (see CLAUDE.md, Inference API boundary). The -harness is a *client* of ``/invocations``, not a new server surface. +(app-api "Run now" routes, the Phase-B dispatcher/worker Lambdas) and must +never be an inference-api route — the AgentCore Runtime data plane only +exposes ``/invocations`` + ``/ping`` (see CLAUDE.md, Inference API boundary). +The harness is a *client* of ``/invocations``, not a new server surface. -Spike status: see docs/specs/harness-entrypoint-spike-findings.md. +Design + spike evidence: docs/specs/harness-entrypoint-spike-findings.md. """ from apis.shared.harness.auth import ( @@ -20,6 +20,11 @@ StaticBearerAuth, ) from apis.shared.harness.governance import GovernanceFloor, RunAuditRecorder +from apis.shared.harness.grants import ( + HEADLESS_GRANT_MAX_AGE_DAYS, + HeadlessGrant, + HeadlessGrantService, +) from apis.shared.harness.models import ( OAuthConsentRequired, RunResult, @@ -32,7 +37,10 @@ "BearerAuthStrategy", "CognitoRefreshBearerAuth", "GovernanceFloor", + "HEADLESS_GRANT_MAX_AGE_DAYS", "HeadlessAuthError", + "HeadlessGrant", + "HeadlessGrantService", "OAuthConsentRequired", "RunAuditRecorder", "RunResult", diff --git a/backend/src/apis/shared/harness/auth.py b/backend/src/apis/shared/harness/auth.py index b7ff4a030..7366195f4 100644 --- a/backend/src/apis/shared/harness/auth.py +++ b/backend/src/apis/shared/harness/auth.py @@ -1,8 +1,9 @@ -"""Unattended bearer minting for headless runs (Unknown 1). +"""Unattended bearer minting for headless runs. The AgentCore Runtime is provisioned with a **Cognito customJwtAuthorizer** (`inference-agentcore-construct.ts`: discovery URL = the user pool, -`allowedClients` = [BFF app client]). Spike probes against dev-ai proved: +`allowedClients` = [BFF app client]). Spike probes against dev-ai proved +(see docs/specs/harness-entrypoint-spike-findings.md, Unknown 1): - A platform **workload access token** (`GetWorkloadAccessTokenForUserId`) is NOT accepted as the `/invocations` bearer — it is an opaque encrypted @@ -12,30 +13,31 @@ is configured: ``AccessDeniedException: Authorization method mismatch``. So the only front door is a real Cognito access token for the owning user, -minted by the platform. :class:`CognitoRefreshBearerAuth` implements the -zero-infra-change path: exchange the user's stored BFF refresh token -(`REFRESH_TOKEN_AUTH` + SECRET_HASH — the exact machinery -`SessionRefreshMiddleware` already uses) for a fresh 1-hour access token. -The pool does not rotate refresh tokens (CDK default), so the mint does not -disturb the user's live browser sessions. +minted by the platform. :class:`CognitoRefreshBearerAuth` implements that: +exchange the refresh token pinned in the user's **headless-grant record** +(`apis.shared.harness.grants` — created when the user enables headless +runs, revocable, TTL-bounded) via `REFRESH_TOKEN_AUTH` + SECRET_HASH — the +exact machinery `SessionRefreshMiddleware` already runs for browser +sessions. + +The minted token then works three layers deep: gateway JWT authorizer → +container `get_current_user_trusted` (`sub` → user_id, so memory/RBAC/ +quota/session all resolve to the right user) → forwarded to forward-auth +MCP servers, which validate `client_id == BFF client`. The workload identity is still essential — but one layer down: *inside* the runtime, connector tokens are minted from the vault via `GetWorkloadAccessTokenForUserId` keyed by the `sub` of the bearer we send (`apis/shared/oauth/agentcore_identity.py`). The front-door bearer and the -vault leg are two different trust boundaries; the spike brief's "try first" -path conflated them. +vault leg are two different trust boundaries. """ from __future__ import annotations -import asyncio import logging -import os from typing import Optional, Protocol -import boto3 - +from apis.shared.harness.grants import HeadlessGrantService from apis.shared.sessions_bff.refresh import CognitoRefreshClient, CognitoRefreshError logger = logging.getLogger(__name__) @@ -44,17 +46,20 @@ class HeadlessAuthError(RuntimeError): """No bearer could be minted for the requested user. - For a scheduled trigger this should pause the schedule (analogous to - KB-sync's ``paused_reauth``) — the user must log in again to renew the - grant. + Raised when the user has no active headless grant (never enabled, + revoked, or expired past the login-recency window) or when Cognito + refuses the refresh exchange. For a scheduled trigger this should pause + the schedule (analogous to KB-sync's ``paused_reauth``) — the user must + log in and re-enable before headless runs can act as them again. """ class BearerAuthStrategy(Protocol): """Seam between the runner and however a bearer is obtained. - Phase A can add strategies (e.g. a dedicated headless-grant record, or - an M2M client + trusted `user_id` payload) without touching the runner. + Additional strategies (e.g. an M2M client + trusted `user_id` payload, + were the runtime's authorizer ever reconfigured) can be added without + touching the runner. """ async def mint_bearer_for_user(self, user_id: str) -> str: ... @@ -71,81 +76,65 @@ async def mint_bearer_for_user(self, user_id: str) -> str: class CognitoRefreshBearerAuth: - """Mint a per-owner access token from the user's stored refresh token. - - Reads the newest BFF session row for ``user_id`` and runs the standard - Cognito refresh exchange. Requires the caller's IAM principal to read - the BFF sessions table and the app-client secret — app-api's role - already holds both grants. - - Spike-scoped caveats (Phase A must address): - - The BFF sessions table is keyed by ``session_id`` only; finding a - user's row is a filtered ``Scan``. Fine at spike scale; Phase A needs - a ``user_id`` GSI or (better) a dedicated headless-grant record that - stores a refresh token minted at opt-in time with its own lifecycle. - - The grant inherits BFF session lifetime (~30-day absolute cap + - row TTL): a user who hasn't logged in recently cannot be run - headlessly. That is arguably the *right* governance default, but it - must be an explicit product decision, not an accident. + """Mint a per-owner access token from the user's headless grant. + + Resolves the newest active :class:`~apis.shared.harness.grants.HeadlessGrant` + for ``user_id`` (a GSI query — no table Scan) and runs the standard + Cognito refresh exchange against its pinned refresh token. Requires the + caller's IAM principal to read/write the BFF sessions table (where + grants live) and the BFF app-client secret — app-api's role already + holds both grants. + + Rotation-aware: if Cognito ever rotates the refresh token during a mint + (rotation is off on this pool today), the replacement is persisted back + onto the grant before the access token is returned, so the grant is + never stranded holding a dead token. """ def __init__( self, *, - sessions_table_name: Optional[str] = None, + grants: Optional[HeadlessGrantService] = None, refresh_client: Optional[CognitoRefreshClient] = None, - region: Optional[str] = None, ) -> None: - self._table_name = sessions_table_name or os.environ.get( - "BFF_SESSIONS_TABLE_NAME", "" - ) + self._grants = grants or HeadlessGrantService() self._refresh_client = refresh_client or CognitoRefreshClient() - self._region = region or os.environ.get("AWS_REGION", "us-west-2") - - def _newest_session_row(self, user_id: str) -> Optional[dict]: - if not self._table_name: - raise HeadlessAuthError("BFF_SESSIONS_TABLE_NAME is not configured") - table = boto3.resource("dynamodb", region_name=self._region).Table( - self._table_name - ) - from boto3.dynamodb.conditions import Attr - - rows: list[dict] = [] - kwargs = {"FilterExpression": Attr("user_id").eq(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: - return None - return max(rows, key=lambda r: int(r.get("last_seen_at") or 0)) async def mint_bearer_for_user(self, user_id: str) -> str: - row = await asyncio.to_thread(self._newest_session_row, user_id) - if row is None: + """Return a fresh Cognito access token for ``user_id``. + + Raises: + HeadlessAuthError: no active grant, or Cognito refused the + refresh exchange (token expired/revoked upstream). + """ + grant = await self._grants.get_active_grant(user_id) + if grant is None: raise HeadlessAuthError( - f"No stored BFF session for user {user_id}; the user must " - "log in before headless runs can act as them." + f"No active headless grant for user {user_id}; the user must " + "log in and enable headless runs before the platform can act " + "as them." ) try: refreshed = await self._refresh_client.refresh( - username=str(row["username"]), - refresh_token=str(row["cognito_refresh_token"]), + username=grant.username, + refresh_token=grant.cognito_refresh_token, ) except CognitoRefreshError as exc: raise HeadlessAuthError( - f"Cognito refused the refresh exchange for user {user_id}: {exc}" + f"Cognito refused the refresh exchange for user {user_id} " + f"(grant {grant.grant_id}): {exc}" ) from exc - if refreshed.refresh_token != str(row["cognito_refresh_token"]): - # Rotation is off on this pool; if it is ever enabled this mint - # would invalidate the user's browser session unless persisted. - logger.warning( + + if refreshed.refresh_token != grant.cognito_refresh_token: + logger.info( "Cognito rotated the refresh token during a headless mint for " - "user %s — the stored BFF session row is now stale. Enable " - "rotation-aware persistence before using this strategy on a " - "rotating pool.", + "user %s; persisting the replacement onto grant %s", user_id, + grant.grant_id, ) + await self._grants.persist_rotated_refresh_token( + grant.grant_id, refreshed.refresh_token + ) + + await self._grants.record_use(grant.grant_id) return refreshed.access_token diff --git a/backend/src/apis/shared/harness/governance.py b/backend/src/apis/shared/harness/governance.py index 13cb04f42..a5abadc94 100644 --- a/backend/src/apis/shared/harness/governance.py +++ b/backend/src/apis/shared/harness/governance.py @@ -2,30 +2,34 @@ The moment an agent turn runs unattended *as a user*, it touches user data with no human in the loop — so every headless run passes through this floor. -The spike implements exactly one slice for real (the audit record); the -guardrails and data-classification checkpoints are explicit no-op seams so -Phase A fills them in without touching the runner's control flow. +The "Run now" phase (PR-1) ships **audit-only + wired seams** (locked +decision, agentic-platform-primitives.md §6): the audit checkpoints are +implemented and fail-closed on start; the guardrails and +data-classification checkpoints are deliberate no-op seams whose +implementations are gated to the *scheduled* (fully unattended) trigger in +a later phase. "Run now" is user-initiated and attended — the human is in +the loop — so the audit trail is the floor it needs. Hook points (called by ``run_agent_headless`` in this order): 1. ``on_run_start`` — AUDIT (implemented): who/what/when/trigger, before - any token is minted or model called. -2. ``check_input`` — GUARDRAILS seam (no-op): Phase A calls Bedrock - ``ApplyGuardrail`` on the prompt; a blocked verdict raises before the - run spends tokens or reads user data. -3. ``classify_output`` — DATA-CLASSIFICATION seam (no-op): Phase A runs the - PII/FERPA checkpoint over the final message (and tool-result previews) - before the result is delivered anywhere. + any token is minted or model called. **Fail-closed**: no audit record → + no run. +2. ``check_input`` — GUARDRAILS seam (no-op): the scheduled-runs phase + calls Bedrock ``ApplyGuardrail`` on the prompt here; a blocked verdict + raises before the run spends tokens or reads user data. +3. ``classify_output`` — DATA-CLASSIFICATION seam (no-op): the + scheduled-runs phase runs the PII/FERPA checkpoint over the final + message (and tool-result previews) here, before delivery. 4. ``on_run_end`` — AUDIT (implemented): outcome, stop reason, tool - names, usage. + names, usage. Best-effort. Audit records are written to the sessions-metadata table under ``PK=USER#{user_id}, SK=RUN#{run_id}``. The session listing queries ``begins_with(SK, 'S#ACTIVE#')`` and the GSI lookups use their own key shapes, so ``RUN#`` items are invisible to every existing access path. -Phase A should promote run-records to their own table (or a documented SK -family) with a "due/recent runs" GSI; for the spike the point is that the -seam exists and every run leaves a durable, queryable trail. +When the scheduler lands (Phase B), promote run-records to their own table +(or a documented SK family) with a "recent runs per user" GSI. """ from __future__ import annotations @@ -150,19 +154,22 @@ async def on_run_start( ) async def check_input(self, *, prompt: str, user_id: str) -> None: - """GUARDRAILS seam — no-op in the spike. + """GUARDRAILS seam — deliberate no-op for attended "Run now". - Phase A: Bedrock ``ApplyGuardrail`` (source=INPUT) on the prompt; - raise a ``GovernanceBlocked`` error on a blocked verdict so the - runner records an audited, non-retryable failure. + Scheduled-runs phase: Bedrock ``ApplyGuardrail`` (source=INPUT) on + the prompt; raise a ``GovernanceBlocked`` error on a blocked + verdict so the runner records an audited, non-retryable failure + before any token is spent. The runner already calls this on every + run, so filling it in requires no control-flow change. """ async def classify_output(self, *, result: RunResult) -> None: - """DATA-CLASSIFICATION seam — no-op in the spike. + """DATA-CLASSIFICATION seam — deliberate no-op for attended "Run now". - Phase A: PII/FERPA checkpoint over ``result.final_message`` and - ``result.tool_trace`` previews before delivery. May redact (mutate - the result) or block (raise), per policy. + Scheduled-runs phase: PII/FERPA checkpoint over + ``result.final_message`` and ``result.tool_trace`` previews before + delivery. May redact (mutate the result) or block (raise), per + policy. The runner already calls this before delivery on every run. """ async def on_run_end(self, *, result: RunResult) -> None: diff --git a/backend/src/apis/shared/harness/grants.py b/backend/src/apis/shared/harness/grants.py new file mode 100644 index 000000000..b9dd62637 --- /dev/null +++ b/backend/src/apis/shared/harness/grants.py @@ -0,0 +1,331 @@ +"""Headless-run grants — the durable "act as me" record for unattended runs. + +A headless run mints a per-owner Cognito access token (see +``apis.shared.harness.auth``). The credential that mint consumes must NOT be +a silently-reused BFF browser session: sessions are an authentication +artifact with an 8-hour sliding TTL and no user-visible lifecycle. Instead, +each user who enables headless runs gets an explicit **headless-grant +record** with its own consent/revocation lifecycle: + +* **Create-on-enable** — when a user turns on the feature (today: the "Run + now" route; later: the schedules SPA), the grant pins the refresh token + from their *live, attended* session. Enabling again re-pins the token and + slides the expiry window. +* **Lookup** — unattended callers resolve the newest active grant by + ``user_id`` via a sparse GSI (a direct query, replacing the spike's + full-table ``Scan``). +* **Revoke** — the user (or an admin) can kill the grant at any time; the + stored refresh token is removed in the same write so a revoked record + retains no usable credential. + +Storage: items live in the BFF sessions table (same data classification — +it already holds Cognito refresh tokens — and the same IAM grants), keyed +``PK=HEADLESS-GRANT#{grant_id}, SK=META`` so they are invisible to every +session access path (sessions key by ``SESSION#{id}``). Only grant items +carry the ``grant_user_id`` attribute, so the ``HeadlessGrantUserIndex`` +GSI (``grant_user_id`` / ``created_at``) is sparse: session rows never +project into it. + +**Login-recency policy (documented product decision):** the platform may +act headlessly as a user only within ``HEADLESS_GRANT_MAX_AGE_DAYS`` +(default **30**, matching the Cognito app client's refresh-token validity — +the CDK default we deploy with) of the login that produced the pinned +token. The grant's DynamoDB TTL is anchored to that login +(``token_issued_at``), so the record expires no later than the token it +wraps. Cognito remains the hard ceiling either way: a refresh exchange +against a token older than the pool's validity fails and surfaces as +``HeadlessAuthError``, which scheduled callers should treat as +"pause until the user logs in again" (the KB-sync ``paused_reauth`` +analog). +""" + +from __future__ import annotations + +import asyncio +import logging +import os +import time +import uuid +from dataclasses import dataclass +from typing import Optional + +import boto3 +from boto3.dynamodb.conditions import Key +from botocore.exceptions import ClientError + +logger = logging.getLogger(__name__) + +#: "Must have logged in within N days" — see the module docstring. Matches +#: the Cognito refresh-token validity (CDK default: 30 days). +HEADLESS_GRANT_MAX_AGE_DAYS = 30 + +GRANT_USER_INDEX_NAME = "HeadlessGrantUserIndex" + +_STATUS_ACTIVE = "active" +_STATUS_REVOKED = "revoked" + + +def _max_age_seconds() -> int: + days = int(os.environ.get("HEADLESS_GRANT_MAX_AGE_DAYS", HEADLESS_GRANT_MAX_AGE_DAYS)) + return days * 24 * 60 * 60 + + +@dataclass +class HeadlessGrant: + """One user's standing consent for the platform to run as them.""" + + grant_id: str + user_id: str + username: str # Cognito username; required for SECRET_HASH on refresh + cognito_refresh_token: str + status: str # "active" | "revoked" + created_at: int # epoch seconds (grant creation) + updated_at: int # epoch seconds (last enable/re-pin or token rotation) + token_issued_at: int # epoch seconds — login that produced the token + ttl: int # epoch seconds; DynamoDB TTL = token_issued_at + max age + last_used_at: Optional[int] = None + revoked_at: Optional[int] = None + + @property + def is_active(self) -> bool: + return self.status == _STATUS_ACTIVE and self.ttl > int(time.time()) + + +class HeadlessGrantService: + """Create-on-enable / lookup / revoke for headless-run grants. + + Async-shaped like ``SessionRepository``: every boto3 round-trip is + offloaded via ``asyncio.to_thread`` so the event loop stays free. + """ + + def __init__( + self, + *, + table_name: Optional[str] = None, + region: Optional[str] = None, + ) -> None: + self._table_name = table_name or os.environ.get("BFF_SESSIONS_TABLE_NAME", "") + self._region = region or os.environ.get("AWS_REGION", "us-west-2") + self._table = None + + @property + def enabled(self) -> bool: + return bool(self._table_name) + + def _get_table(self): + if self._table is None: + if not self._table_name: + raise RuntimeError( + "BFF_SESSIONS_TABLE_NAME is required for headless grants" + ) + self._table = boto3.resource( + "dynamodb", region_name=self._region + ).Table(self._table_name) + return self._table + + @staticmethod + def _key(grant_id: str) -> dict: + return {"PK": f"HEADLESS-GRANT#{grant_id}", "SK": "META"} + + @staticmethod + def _item_to_grant(item: dict) -> HeadlessGrant: + return HeadlessGrant( + grant_id=item["grant_id"], + user_id=item["grant_user_id"], + username=item["username"], + cognito_refresh_token=item.get("cognito_refresh_token", ""), + status=item["status"], + created_at=int(item["created_at"]), + updated_at=int(item["updated_at"]), + token_issued_at=int(item["token_issued_at"]), + ttl=int(item["ttl"]), + last_used_at=int(item["last_used_at"]) if "last_used_at" in item else None, + revoked_at=int(item["revoked_at"]) if "revoked_at" in item else None, + ) + + def _query_grants_sync(self, user_id: str) -> list[dict]: + """Newest-first grant items for a user via the sparse GSI.""" + table = self._get_table() + items: list[dict] = [] + kwargs: dict = { + "IndexName": GRANT_USER_INDEX_NAME, + "KeyConditionExpression": Key("grant_user_id").eq(user_id), + "ScanIndexForward": False, + } + while True: + page = table.query(**kwargs) + items.extend(page.get("Items", [])) + if "LastEvaluatedKey" not in page: + break + kwargs["ExclusiveStartKey"] = page["LastEvaluatedKey"] + return items + + async def get_active_grant(self, user_id: str) -> Optional[HeadlessGrant]: + """Return the newest active, unexpired grant for ``user_id``. + + Expiry is checked application-side too (DynamoDB TTL eviction is + best-effort, not real-time — same defense-in-depth as the session + repository). + """ + items = await asyncio.to_thread(self._query_grants_sync, user_id) + for item in items: + grant = self._item_to_grant(item) + if grant.is_active: + return grant + return None + + async def enable( + self, + *, + user_id: str, + username: str, + refresh_token: str, + token_issued_at: Optional[int] = None, + ) -> HeadlessGrant: + """Create (or renew) the user's grant from an attended session. + + Callers pass the refresh token from the user's *live* BFF session + plus that session's ``created_at`` as ``token_issued_at`` — the + login instant anchors the grant's TTL, because Cognito's + refresh-token validity runs from token issuance, not from when we + pin it. Re-enabling re-pins the token onto the existing grant + (stable ``grant_id`` for audit continuity) and slides the window. + """ + now = int(time.time()) + issued_at = token_issued_at or now + ttl = issued_at + _max_age_seconds() + + existing = await self.get_active_grant(user_id) + if existing is not None: + def _renew() -> None: + self._get_table().update_item( + Key=self._key(existing.grant_id), + UpdateExpression=( + "SET cognito_refresh_token = :rt, updated_at = :now, " + "token_issued_at = :iss, #ttl = :ttl" + ), + ExpressionAttributeNames={"#ttl": "ttl"}, + ExpressionAttributeValues={ + ":rt": refresh_token, + ":now": now, + ":iss": issued_at, + ":ttl": ttl, + }, + ) + + await asyncio.to_thread(_renew) + logger.info( + "Renewed headless grant %s for user %s", existing.grant_id, user_id + ) + existing.cognito_refresh_token = refresh_token + existing.updated_at = now + existing.token_issued_at = issued_at + existing.ttl = ttl + return existing + + grant = HeadlessGrant( + grant_id=f"hlg-{uuid.uuid4().hex[:12]}", + user_id=user_id, + username=username, + cognito_refresh_token=refresh_token, + status=_STATUS_ACTIVE, + created_at=now, + updated_at=now, + token_issued_at=issued_at, + ttl=ttl, + ) + + def _put() -> None: + self._get_table().put_item( + Item={ + **self._key(grant.grant_id), + "grant_id": grant.grant_id, + "grant_user_id": grant.user_id, + "username": grant.username, + "cognito_refresh_token": grant.cognito_refresh_token, + "status": grant.status, + "created_at": grant.created_at, + "updated_at": grant.updated_at, + "token_issued_at": grant.token_issued_at, + "ttl": grant.ttl, + } + ) + + await asyncio.to_thread(_put) + logger.info("Created headless grant %s for user %s", grant.grant_id, user_id) + return grant + + async def revoke(self, user_id: str) -> bool: + """Revoke every active grant for ``user_id``. + + The stored refresh token is REMOVEd in the same write — a revoked + record keeps its audit fields but no usable credential. Returns + True iff at least one grant was revoked. + """ + items = await asyncio.to_thread(self._query_grants_sync, user_id) + now = int(time.time()) + revoked_any = False + for item in items: + if item.get("status") != _STATUS_ACTIVE: + continue + grant_id = str(item["grant_id"]) + + def _revoke(gid: str = grant_id) -> None: + self._get_table().update_item( + Key=self._key(gid), + UpdateExpression=( + "SET #s = :revoked, revoked_at = :now, updated_at = :now " + "REMOVE cognito_refresh_token" + ), + ExpressionAttributeNames={"#s": "status"}, + ExpressionAttributeValues={":revoked": _STATUS_REVOKED, ":now": now}, + ) + + await asyncio.to_thread(_revoke) + logger.info("Revoked headless grant %s for user %s", grant_id, user_id) + revoked_any = True + return revoked_any + + async def persist_rotated_refresh_token( + self, grant_id: str, refresh_token: str + ) -> None: + """Persist a rotated refresh token back onto the grant. + + The pool we deploy does not rotate refresh tokens today, but if + rotation is ever enabled the old token dies the moment Cognito + rotates it — failing to persist the replacement would strand the + grant after one headless mint. Callers treat failures as fatal for + the mint (better to fail loudly than to silently burn the grant's + last valid token). + """ + + def _update() -> None: + self._get_table().update_item( + Key=self._key(grant_id), + UpdateExpression=( + "SET cognito_refresh_token = :rt, updated_at = :now" + ), + ConditionExpression="attribute_exists(PK)", + ExpressionAttributeValues={ + ":rt": refresh_token, + ":now": int(time.time()), + }, + ) + + await asyncio.to_thread(_update) + + async def record_use(self, grant_id: str) -> None: + """Best-effort ``last_used_at`` touch — never fails a run.""" + + def _touch() -> None: + self._get_table().update_item( + Key=self._key(grant_id), + UpdateExpression="SET last_used_at = :now", + ConditionExpression="attribute_exists(PK)", + ExpressionAttributeValues={":now": int(time.time())}, + ) + + try: + await asyncio.to_thread(_touch) + except (ClientError, RuntimeError) as exc: + logger.warning("Headless grant %s last-used touch failed: %s", grant_id, exc) diff --git a/backend/src/apis/shared/harness/runner.py b/backend/src/apis/shared/harness/runner.py index 19412d10e..5b637c17e 100644 --- a/backend/src/apis/shared/harness/runner.py +++ b/backend/src/apis/shared/harness/runner.py @@ -1,4 +1,4 @@ -"""`run_agent_headless` — the F1 headless run entrypoint (spike). +"""`run_agent_headless` — the F1 headless run entrypoint. Given ``(user_id, prompt)`` and no live session: mint a per-owner bearer, POST the AgentCore Runtime ``/invocations`` data plane, drain the SSE stream @@ -9,9 +9,9 @@ A2A-readiness: the seam is ``(user_id, prompt, resolved config) → RunResult`` plus the optional ``on_event`` callback, which relays every typed SSE event as it arrives — an A2A server front can map that stream onto task status -updates -without a rewrite. Reminder from CLAUDE.md: if this is ever exposed as an -A2A server, its advertised ``capabilities`` MUST include ``streaming=True``. +updates without a rewrite. Reminder from CLAUDE.md: if this is ever exposed +as an A2A server, its advertised ``capabilities`` MUST include +``streaming=True``. """ from __future__ import annotations @@ -39,6 +39,16 @@ OnEvent = Callable[[str, Dict[str, Any]], Awaitable[None]] +def _build_http_client(timeout_seconds: float) -> httpx.AsyncClient: + """Single seam where the runner's upstream client is constructed. + + Tests substitute a MockTransport-backed client here without patching + the global ``httpx.AsyncClient`` symbol (same pattern as the chat + proxy's ``_build_upstream_client``). + """ + return httpx.AsyncClient(timeout=httpx.Timeout(timeout_seconds)) + + def _now_iso() -> str: return datetime.now(timezone.utc).isoformat().replace("+00:00", "Z") @@ -46,12 +56,20 @@ def _now_iso() -> str: def build_invocations_url(base_url: str) -> str: """Resolve the upstream `/invocations` URL from an INFERENCE_API_URL-style base. - Same resolution as the app-api chat proxy (`proxy_routes.py`): in cloud - the base is the AgentCore data plane - (`https://bedrock-agentcore..amazonaws.com/runtimes/`) whose - ARN path segment must be percent-encoded and given a qualifier; locally - it's a plain FastAPI origin. Phase A should point the proxy at this - shared copy rather than keeping two. + This is the single canonical resolver — the app-api chat proxy + (`proxy_routes.py`) and the MCP Apps proxy import it from here. + + Cloud: the base is the AgentCore Runtime data plane + (`https://bedrock-agentcore..amazonaws.com/runtimes/`), + where `` is unencoded in SSM. The data-plane route is + `POST /runtimes/{agentRuntimeArn}/invocations?qualifier={qualifier}` + with `{agentRuntimeArn}` as a single URL-encoded path segment — the + ARN's literal `/` and `:` must be percent-encoded or AWS returns 404. + A `qualifier` is also required; we use `DEFAULT`. + + Local dev: the base 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/" @@ -91,6 +109,14 @@ async def run_agent_headless( existing ``InvocationRequest`` contract — the entrypoint resolves *no* new config type (scheduled-agent-runs.md decision #6). + ``enabled_tools=None`` means "the user's defaults" — inference-api + resolves it to **all RBAC-allowed tools** for the owner, exactly as an + attended chat turn would. That is the right semantic for the attended + "Run now" surface; *schedules* should instead snapshot an explicit tool + list at creation time so a sleeping schedule's behavior doesn't shift + when the catalog or the owner's grants change underneath it + (spike-findings punch list #7 — enforce in the Phase B schedule model). + Returns a ``RunResult`` in all outcomes except audit-start failure and ``HeadlessAuthError`` (both raise: a run we cannot audit or authenticate must not execute). @@ -147,9 +173,7 @@ async def run_agent_headless( acc = InvocationStreamAccumulator() try: - async with httpx.AsyncClient( - timeout=httpx.Timeout(timeout_seconds) - ) as client: + async with _build_http_client(timeout_seconds) as client: async with client.stream( "POST", url, diff --git a/backend/tests/apis/shared/test_harness_grants.py b/backend/tests/apis/shared/test_harness_grants.py new file mode 100644 index 000000000..ae267ef2c --- /dev/null +++ b/backend/tests/apis/shared/test_harness_grants.py @@ -0,0 +1,319 @@ +"""Tests for the headless-grant record + grant-backed bearer minting. + +The grant record is the durable "act as me" consent for headless runs +(``apis/shared/harness/grants.py``): create-on-enable from an attended +session, per-owner lookup via the sparse ``HeadlessGrantUserIndex`` GSI +(replacing the spike's BFF-table Scan), and total revocation (the stored +refresh token is deleted in the revoke write). +""" + +from __future__ import annotations + +import time + +import pytest +from botocore.exceptions import ClientError + +from apis.shared.harness.auth import CognitoRefreshBearerAuth, HeadlessAuthError +from apis.shared.harness.grants import ( + HEADLESS_GRANT_MAX_AGE_DAYS, + HeadlessGrant, + HeadlessGrantService, +) +from apis.shared.sessions_bff.refresh import CognitoRefreshError, RefreshResult + +NOW = int(time.time()) +MAX_AGE_SECONDS = HEADLESS_GRANT_MAX_AGE_DAYS * 24 * 60 * 60 + + +class FakeTable: + """Duck-typed DynamoDB Table capturing writes; query returns canned pages.""" + + def __init__(self, query_items=None): + self.query_items = list(query_items or []) + self.put_items: list[dict] = [] + self.update_calls: list[dict] = [] + self.query_kwargs: dict = {} + self.update_error: Exception | None = None + + def query(self, **kwargs): + self.query_kwargs = kwargs + return {"Items": list(self.query_items)} + + def put_item(self, Item): + self.put_items.append(Item) + # Newest-first, as the descending GSI query would return it. + self.query_items.insert(0, Item) + + def update_item(self, **kwargs): + if self.update_error is not None: + raise self.update_error + self.update_calls.append(kwargs) + + +def _service(table: FakeTable) -> HeadlessGrantService: + service = HeadlessGrantService(table_name="fake-table") + service._table = table + return service + + +def _grant_item( + *, + grant_id: str = "hlg-abc", + user_id: str = "user-1", + status: str = "active", + created_at: int = NOW - 100, + ttl: int = NOW + 1000, +) -> dict: + return { + "PK": f"HEADLESS-GRANT#{grant_id}", + "SK": "META", + "grant_id": grant_id, + "grant_user_id": user_id, + "username": "user1", + "cognito_refresh_token": "rt-stored", + "status": status, + "created_at": created_at, + "updated_at": created_at, + "token_issued_at": created_at, + "ttl": ttl, + } + + +# --------------------------------------------------------------------------- +# HeadlessGrantService +# --------------------------------------------------------------------------- + + +class TestEnable: + @pytest.mark.asyncio + async def test_creates_grant_with_sparse_gsi_key_and_login_anchored_ttl(self): + table = FakeTable() + service = _service(table) + issued_at = NOW - 3600 # the login that produced the token + + grant = await service.enable( + user_id="user-1", + username="user1", + refresh_token="rt-1", + token_issued_at=issued_at, + ) + + assert grant.grant_id.startswith("hlg-") + item = table.put_items[0] + assert item["PK"] == f"HEADLESS-GRANT#{grant.grant_id}" + assert item["SK"] == "META" + # Sparse GSI partition key — only grant items carry it. + assert item["grant_user_id"] == "user-1" + assert item["cognito_refresh_token"] == "rt-1" + assert item["status"] == "active" + # "Must have logged in within N days": TTL anchors to the login. + assert item["ttl"] == issued_at + MAX_AGE_SECONDS + + @pytest.mark.asyncio + async def test_renews_existing_active_grant_in_place(self): + table = FakeTable([_grant_item()]) + service = _service(table) + + grant = await service.enable( + user_id="user-1", username="user1", refresh_token="rt-new" + ) + + assert grant.grant_id == "hlg-abc" # stable id for audit continuity + assert not table.put_items # renew, not a second record + (call,) = table.update_calls + assert call["Key"] == {"PK": "HEADLESS-GRANT#hlg-abc", "SK": "META"} + assert call["ExpressionAttributeValues"][":rt"] == "rt-new" + assert grant.cognito_refresh_token == "rt-new" + + @pytest.mark.asyncio + async def test_max_age_days_is_env_overridable(self, monkeypatch): + monkeypatch.setenv("HEADLESS_GRANT_MAX_AGE_DAYS", "7") + table = FakeTable() + service = _service(table) + + await service.enable( + user_id="user-1", + username="user1", + refresh_token="rt-1", + token_issued_at=NOW, + ) + + assert table.put_items[0]["ttl"] == NOW + 7 * 24 * 60 * 60 + + +class TestGetActiveGrant: + @pytest.mark.asyncio + async def test_returns_newest_active_grant(self): + table = FakeTable([_grant_item()]) + service = _service(table) + + grant = await service.get_active_grant("user-1") + + assert isinstance(grant, HeadlessGrant) + assert grant.grant_id == "hlg-abc" + assert grant.is_active + # The lookup is a GSI query, not a Scan. + assert table.query_kwargs["IndexName"] == "HeadlessGrantUserIndex" + assert table.query_kwargs["ScanIndexForward"] is False + + @pytest.mark.asyncio + async def test_skips_revoked_and_expired_grants(self): + table = FakeTable( + [ + _grant_item(grant_id="hlg-revoked", status="revoked"), + # TTL passed but DynamoDB hasn't swept yet — defense in depth. + _grant_item(grant_id="hlg-expired", ttl=NOW - 5), + _grant_item(grant_id="hlg-live", created_at=NOW - 999), + ] + ) + service = _service(table) + + grant = await service.get_active_grant("user-1") + + assert grant is not None and grant.grant_id == "hlg-live" + + @pytest.mark.asyncio + async def test_returns_none_when_user_has_no_grants(self): + service = _service(FakeTable()) + assert await service.get_active_grant("user-1") is None + + +class TestRevoke: + @pytest.mark.asyncio + async def test_revoke_removes_the_stored_credential(self): + table = FakeTable([_grant_item()]) + service = _service(table) + + assert await service.revoke("user-1") is True + (call,) = table.update_calls + assert "REMOVE cognito_refresh_token" in call["UpdateExpression"] + assert call["ExpressionAttributeValues"][":revoked"] == "revoked" + + @pytest.mark.asyncio + async def test_revoke_is_false_when_nothing_active(self): + table = FakeTable([_grant_item(status="revoked")]) + service = _service(table) + + assert await service.revoke("user-1") is False + assert not table.update_calls + + +class TestRecordUse: + @pytest.mark.asyncio + async def test_touch_failure_never_raises(self): + table = FakeTable() + table.update_error = ClientError( + {"Error": {"Code": "ProvisionedThroughputExceededException"}}, + "UpdateItem", + ) + service = _service(table) + + await service.record_use("hlg-abc") # must not raise + + +# --------------------------------------------------------------------------- +# CognitoRefreshBearerAuth (grant-backed mint) +# --------------------------------------------------------------------------- + + +class FakeGrants: + def __init__(self, grant: HeadlessGrant | None): + self.grant = grant + self.persisted: list[tuple[str, str]] = [] + self.used: list[str] = [] + + async def get_active_grant(self, user_id: str): + return self.grant + + async def persist_rotated_refresh_token(self, grant_id: str, refresh_token: str): + self.persisted.append((grant_id, refresh_token)) + + async def record_use(self, grant_id: str): + self.used.append(grant_id) + + +class FakeRefreshClient: + def __init__(self, result: RefreshResult | None = None, error: Exception | None = None): + self.result = result + self.error = error + self.calls: list[dict] = [] + + async def refresh(self, *, username: str, refresh_token: str) -> RefreshResult: + self.calls.append({"username": username, "refresh_token": refresh_token}) + if self.error is not None: + raise self.error + assert self.result is not None + return self.result + + +def _live_grant() -> HeadlessGrant: + return HeadlessGrant( + grant_id="hlg-abc", + user_id="user-1", + username="user1", + cognito_refresh_token="rt-stored", + status="active", + created_at=NOW - 100, + updated_at=NOW - 100, + token_issued_at=NOW - 100, + ttl=NOW + 1000, + ) + + +class TestCognitoRefreshBearerAuth: + @pytest.mark.asyncio + async def test_mints_from_the_grant_and_records_use(self): + grants = FakeGrants(_live_grant()) + refresh = FakeRefreshClient( + RefreshResult( + access_token="at-fresh", + refresh_token="rt-stored", # no rotation on this pool + id_token=None, + access_token_exp=NOW + 3600, + ) + ) + auth = CognitoRefreshBearerAuth(grants=grants, refresh_client=refresh) + + token = await auth.mint_bearer_for_user("user-1") + + assert token == "at-fresh" + assert refresh.calls == [{"username": "user1", "refresh_token": "rt-stored"}] + assert grants.used == ["hlg-abc"] + assert grants.persisted == [] + + @pytest.mark.asyncio + async def test_no_active_grant_raises_headless_auth_error(self): + auth = CognitoRefreshBearerAuth( + grants=FakeGrants(None), refresh_client=FakeRefreshClient() + ) + + with pytest.raises(HeadlessAuthError, match="No active headless grant"): + await auth.mint_bearer_for_user("user-1") + + @pytest.mark.asyncio + async def test_cognito_refusal_raises_headless_auth_error(self): + auth = CognitoRefreshBearerAuth( + grants=FakeGrants(_live_grant()), + refresh_client=FakeRefreshClient(error=CognitoRefreshError("revoked")), + ) + + with pytest.raises(HeadlessAuthError, match="refused the refresh exchange"): + await auth.mint_bearer_for_user("user-1") + + @pytest.mark.asyncio + async def test_rotated_refresh_token_is_persisted_onto_the_grant(self): + grants = FakeGrants(_live_grant()) + refresh = FakeRefreshClient( + RefreshResult( + access_token="at-fresh", + refresh_token="rt-rotated", + id_token=None, + access_token_exp=NOW + 3600, + ) + ) + auth = CognitoRefreshBearerAuth(grants=grants, refresh_client=refresh) + + await auth.mint_bearer_for_user("user-1") + + assert grants.persisted == [("hlg-abc", "rt-rotated")] diff --git a/backend/tests/apis/shared/test_harness_runner.py b/backend/tests/apis/shared/test_harness_runner.py new file mode 100644 index 000000000..bfc3e6645 --- /dev/null +++ b/backend/tests/apis/shared/test_harness_runner.py @@ -0,0 +1,272 @@ +"""Tests for ``run_agent_headless`` — governance ordering, outcomes, delivery. + +The runner is exercised end-to-end against an ``httpx.MockTransport`` +standing in for the AgentCore Runtime data plane (via the +``_build_http_client`` seam), with a recording governance floor and spied +delivery functions. The key invariant pinned here is the F6a fail-closed +rule: **no audit record → no run** (no bearer mint, no HTTP). +""" + +from __future__ import annotations + +import httpx +import pytest + +import apis.shared.sessions.metadata as sessions_metadata +from apis.shared.harness import runner as runner_module +from apis.shared.harness.auth import HeadlessAuthError, StaticBearerAuth +from apis.shared.harness.governance import GovernanceFloor +from apis.shared.harness.runner import build_invocations_url, run_agent_headless + + +class RecordingAudit: + """Stands in for RunAuditRecorder; optionally fails the start write.""" + + def __init__(self, *, fail_start: bool = False): + self.fail_start = fail_start + self.starts: list[dict] = [] + self.ends: list = [] + + def record_start(self, **kwargs): + if self.fail_start: + raise RuntimeError("dynamo down") + self.starts.append(kwargs) + + def record_end(self, *, result): + self.ends.append(result) + + +class SpyBearerAuth(StaticBearerAuth): + def __init__(self, token: str = "bearer-1"): + super().__init__(token) + self.minted_for: list[str] = [] + + async def mint_bearer_for_user(self, user_id: str) -> str: + self.minted_for.append(user_id) + return await super().mint_bearer_for_user(user_id) + + +class FailingBearerAuth: + async def mint_bearer_for_user(self, user_id: str) -> str: + raise HeadlessAuthError("no grant") + + +def _sse(*events: tuple[str, str]) -> bytes: + return "".join(f"event: {name}\ndata: {data}\n\n" for name, data in events).encode() + + +_HAPPY_STREAM = _sse( + ("message_start", '{"role": "assistant"}'), + ("content_block_delta", '{"contentBlockIndex": 0, "type": "text", "text": "pong"}'), + ("message_stop", '{"stopReason": "end_turn"}'), + ("metadata", '{"usage": {"inputTokens": 5, "outputTokens": 1, "totalTokens": 6}}'), + ("done", "{}"), +) + + +@pytest.fixture +def delivery_spy(monkeypatch): + """Spy the runner's delivery calls (imported lazily from sessions.metadata).""" + calls = {"ensure": [], "title": []} + + async def fake_ensure(session_id, user_id): + calls["ensure"].append((session_id, user_id)) + return True + + async def fake_title(session_id, user_id, title): + calls["title"].append((session_id, user_id, title)) + + monkeypatch.setattr(sessions_metadata, "ensure_session_metadata_exists", fake_ensure) + monkeypatch.setattr(sessions_metadata, "update_session_title", fake_title) + return calls + + +def _mock_http(monkeypatch, handler): + def factory(timeout_seconds: float) -> httpx.AsyncClient: + return httpx.AsyncClient(transport=httpx.MockTransport(handler)) + + monkeypatch.setattr(runner_module, "_build_http_client", factory) + + +# --------------------------------------------------------------------------- +# F6a — audit fail-closed +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_audit_start_failure_fails_closed_before_any_spend(monkeypatch): + """No audit record → no run: neither the bearer mint nor HTTP happens.""" + auth = SpyBearerAuth() + + def no_http(timeout_seconds): # any HTTP attempt is a test failure + raise AssertionError("HTTP client built despite failed audit") + + monkeypatch.setattr(runner_module, "_build_http_client", no_http) + + with pytest.raises(RuntimeError, match="dynamo down"): + await run_agent_headless( + user_id="user-1", + prompt="hi", + auth=auth, + governance=GovernanceFloor(audit=RecordingAudit(fail_start=True)), + ) + + assert auth.minted_for == [] + + +@pytest.mark.asyncio +async def test_auth_failure_still_writes_the_end_audit_record(monkeypatch, delivery_spy): + audit = RecordingAudit() + + with pytest.raises(HeadlessAuthError): + await run_agent_headless( + user_id="user-1", + prompt="hi", + auth=FailingBearerAuth(), + governance=GovernanceFloor(audit=audit), + ) + + assert len(audit.starts) == 1 + (end,) = audit.ends + assert end.status == "error" + assert "auth" in (end.error or "") + + +# --------------------------------------------------------------------------- +# Outcomes +# --------------------------------------------------------------------------- + + +@pytest.mark.asyncio +async def test_happy_path_completes_and_delivers(monkeypatch, delivery_spy): + seen = {} + + def handler(request: httpx.Request) -> httpx.Response: + seen["auth"] = request.headers.get("authorization") + seen["url"] = str(request.url) + return httpx.Response( + 200, + headers={"content-type": "text/event-stream"}, + content=_HAPPY_STREAM, + ) + + _mock_http(monkeypatch, handler) + audit = RecordingAudit() + + result = await run_agent_headless( + user_id="user-1", + prompt="ping", + auth=SpyBearerAuth("bearer-xyz"), + title="My Briefing", + trigger="run_now", + invocations_base_url="http://localhost:8001", + governance=GovernanceFloor(audit=audit), + ) + + assert result.status == "completed" + assert result.final_message == "pong" + assert result.title == "My Briefing" + assert result.usage["usage"]["totalTokens"] == 6 + assert seen["auth"] == "Bearer bearer-xyz" + assert seen["url"] == "http://localhost:8001/invocations" + + # Audit trail: start before, end after, same run id. + assert audit.starts[0]["run_id"] == result.run_id + assert audit.starts[0]["trigger"] == "run_now" + assert audit.ends[0].status == "completed" + + # Delivery: idempotent session ensure + explicit title override. + assert delivery_spy["ensure"] == [(result.session_id, "user-1")] + assert delivery_spy["title"] == [(result.session_id, "user-1", "My Briefing")] + + +@pytest.mark.asyncio +async def test_http_error_from_the_gateway_is_an_error_result(monkeypatch, delivery_spy): + def handler(request: httpx.Request) -> httpx.Response: + return httpx.Response(403, json={"message": "OAuth authorization failed"}) + + _mock_http(monkeypatch, handler) + + result = await run_agent_headless( + user_id="user-1", + prompt="ping", + auth=StaticBearerAuth("t"), + invocations_base_url="http://localhost:8001", + governance=GovernanceFloor(audit=RecordingAudit()), + ) + + assert result.status == "error" + assert "HTTP 403" in (result.error or "") + + +@pytest.mark.asyncio +async def test_stream_without_done_event_is_an_error(monkeypatch, delivery_spy): + def handler(request: httpx.Request) -> httpx.Response: + return httpx.Response( + 200, + headers={"content-type": "text/event-stream"}, + content=_sse(("message_start", '{"role": "assistant"}')), + ) + + _mock_http(monkeypatch, handler) + + result = await run_agent_headless( + user_id="user-1", + prompt="ping", + auth=StaticBearerAuth("t"), + invocations_base_url="http://localhost:8001", + governance=GovernanceFloor(audit=RecordingAudit()), + ) + + assert result.status == "error" + assert "without a done event" in (result.error or "") + + +@pytest.mark.asyncio +async def test_timeout_surfaces_as_timeout_status(monkeypatch, delivery_spy): + def handler(request: httpx.Request) -> httpx.Response: + raise httpx.ReadTimeout("too slow") + + _mock_http(monkeypatch, handler) + + result = await run_agent_headless( + user_id="user-1", + prompt="ping", + auth=StaticBearerAuth("t"), + invocations_base_url="http://localhost:8001", + timeout_seconds=1.0, + governance=GovernanceFloor(audit=RecordingAudit()), + ) + + assert result.status == "timeout" + + +# --------------------------------------------------------------------------- +# build_invocations_url — the single shared resolver (chat proxy imports it) +# --------------------------------------------------------------------------- + + +def test_build_invocations_url_encodes_the_runtime_arn(): + base = ( + "https://bedrock-agentcore.us-west-2.amazonaws.com/runtimes/" + "arn:aws:bedrock-agentcore:us-west-2:123456789012:runtime/my-runtime" + ) + url = build_invocations_url(base) + assert url == ( + "https://bedrock-agentcore.us-west-2.amazonaws.com/runtimes/" + "arn%3Aaws%3Abedrock-agentcore%3Aus-west-2%3A123456789012%3Aruntime%2F" + "my-runtime/invocations?qualifier=DEFAULT" + ) + + +def test_build_invocations_url_local_passthrough(): + assert ( + build_invocations_url("http://localhost:8001") + == "http://localhost:8001/invocations" + ) + + +def test_chat_proxy_uses_the_shared_resolver(): + from apis.app_api.chat import proxy_routes + + assert proxy_routes._build_invocations_url is build_invocations_url From 3137cad531dfb007bde47d56f727b9c856788289 Mon Sep 17 00:00:00 2001 From: Phil Merrell Date: Sun, 5 Jul 2026 18:11:12 -0600 Subject: [PATCH 3/4] feat(runs): cookie-authed "Run now" surface behind flag + RBAC capability MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit POST /runs/now executes one agent turn through the exact unattended path a scheduled run will use (create-on-enable grant -> per-owner Cognito mint -> runtime /invocations -> server-side SSE drain -> governance floor -> session materialization) — the PR-1 validation surface from docs/specs/scheduled-agent-runs.md §7. GET/DELETE /runs/grant expose grant status and total revocation. Gating is two independent controls (spec §6): the SCHEDULED_RUNS_ENABLED kill switch (default ON; only the literal "false" disables — empty workflow vars can't dark-stop prod) and a new `scheduled-runs` RBAC capability resolved through the mature tools grant axis (apis/shared/rbac/capabilities.py) — GA = grant the id to the default role. Auth is the standard SPA cookie dependency per the CLAUDE.md app-api rule; mint failures surface as 409, never 401, so the SPA is not bounced through the login redirect. Co-Authored-By: Claude Opus 4.8 --- backend/src/apis/app_api/main.py | 2 + backend/src/apis/app_api/runs/__init__.py | 1 + backend/src/apis/app_api/runs/routes.py | 304 +++++++++++++++ backend/src/apis/shared/feature_flags.py | 28 +- backend/src/apis/shared/rbac/capabilities.py | 44 +++ .../tests/apis/app_api/test_runs_routes.py | 345 ++++++++++++++++++ 6 files changed, 720 insertions(+), 4 deletions(-) create mode 100644 backend/src/apis/app_api/runs/__init__.py create mode 100644 backend/src/apis/app_api/runs/routes.py create mode 100644 backend/src/apis/shared/rbac/capabilities.py create mode 100644 backend/tests/apis/app_api/test_runs_routes.py diff --git a/backend/src/apis/app_api/main.py b/backend/src/apis/app_api/main.py index 571e3aea7..48af426f5 100644 --- a/backend/src/apis/app_api/main.py +++ b/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": diff --git a/backend/src/apis/app_api/runs/__init__.py b/backend/src/apis/app_api/runs/__init__.py new file mode 100644 index 000000000..194d15f3f --- /dev/null +++ b/backend/src/apis/app_api/runs/__init__.py @@ -0,0 +1 @@ +"""User-facing headless-run surface ("Run now" + headless-grant lifecycle).""" diff --git a/backend/src/apis/app_api/runs/routes.py b/backend/src/apis/app_api/runs/routes.py new file mode 100644 index 000000000..914d1f517 --- /dev/null +++ b/backend/src/apis/app_api/runs/routes.py @@ -0,0 +1,304 @@ +""""Run now" — the attended validation surface for the headless harness. + +``POST /runs/now`` executes one agent turn *through the exact machinery a +scheduled run will use* (headless grant → per-owner Cognito mint → runtime +``/invocations`` → server-side SSE drain → governance floor → session +materialization) while the user is present to watch it. It deliberately +does NOT shortcut through the caller's live access token: the point of the +surface is to validate the unattended path end-to-end (scheduled-runs PR-1, +docs/specs/scheduled-agent-runs.md §7). + +Gating — two independent controls (spec §6): + +* ``SCHEDULED_RUNS_ENABLED`` — per-environment kill switch (default on). + Off → every route here 404s, as if unmounted. +* ``scheduled-runs`` RBAC capability — *who* may use the surface. Granted + to the beta cohort's AppRole; missing → 403. GA = grant to ``default``. + +Auth is the standard SPA cookie dependency (``get_current_user_from_session``) +per the CLAUDE.md app-api rule. The headless grant is **created-on-enable**: +each attended ``POST /runs/now`` pins the caller's live session refresh +token into their grant record (renewing the 30-day login-recency window); +``GET/DELETE /runs/grant`` expose status and revocation. +""" + +from __future__ import annotations + +import logging +from typing import Any, Dict, List, Optional + +from fastapi import APIRouter, Depends, HTTPException, Request, status +from pydantic import BaseModel, Field + +from apis.shared.auth.dependencies import get_current_user_from_session +from apis.shared.auth.models import User +from apis.shared.feature_flags import scheduled_runs_enabled +from apis.shared.harness import ( + CognitoRefreshBearerAuth, + HeadlessAuthError, + HeadlessGrant, + HeadlessGrantService, + RunResult, + run_agent_headless, +) +from apis.shared.rbac.capabilities import ( + SCHEDULED_RUNS_CAPABILITY, + user_has_capability, +) + +logger = logging.getLogger(__name__) + +router = APIRouter(prefix="/runs", tags=["runs"]) + +_MAX_PROMPT_CHARS = 20_000 + +_grant_service: Optional[HeadlessGrantService] = None + + +def get_headless_grant_service() -> HeadlessGrantService: + """Lazy module singleton; tests monkeypatch this factory.""" + global _grant_service + if _grant_service is None: + _grant_service = HeadlessGrantService() + return _grant_service + + +async def require_scheduled_runs_user( + user: User = Depends(get_current_user_from_session), +) -> User: + """Cookie auth + kill switch + cohort capability, in that order. + + 404 when the environment kill switch is off (the surface behaves as if + unmounted — runtime-checked so tests and env flips need no module + reload), 403 when the authenticated caller lacks the ``scheduled-runs`` + capability. + """ + if not scheduled_runs_enabled(): + raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Not found") + if not await user_has_capability(user, SCHEDULED_RUNS_CAPABILITY): + raise HTTPException( + status_code=status.HTTP_403_FORBIDDEN, + detail="You do not have access to scheduled runs.", + ) + return user + + +# ─── API models ──────────────────────────────────────────────────────────── + + +class RunNowRequest(BaseModel): + """Run-config mirrors ``InvocationRequest`` — no new config type.""" + + prompt: str = Field(..., min_length=1, max_length=_MAX_PROMPT_CHARS) + title: Optional[str] = Field(None, max_length=200) + model_id: Optional[str] = Field(None, alias="modelId") + rag_assistant_id: Optional[str] = Field(None, alias="ragAssistantId") + # None = the user's defaults (all RBAC-allowed tools), exactly as an + # attended chat turn resolves them — see run_agent_headless docstring. + enabled_tools: Optional[List[str]] = Field(None, alias="enabledTools") + agent_type: Optional[str] = Field(None, alias="agentType") + + model_config = {"populate_by_name": True} + + +class ToolTraceEntryResponse(BaseModel): + tool_use_id: str = Field(..., alias="toolUseId") + name: str + input: Dict[str, Any] = Field(default_factory=dict) + result_preview: Optional[str] = Field(None, alias="resultPreview") + is_error: bool = Field(False, alias="isError") + + model_config = {"populate_by_name": True} + + +class OAuthConsentRequiredResponse(BaseModel): + provider_id: str = Field(..., alias="providerId") + authorization_url: str = Field(..., alias="authorizationUrl") + + model_config = {"populate_by_name": True} + + +class RunNowResponse(BaseModel): + run_id: str = Field(..., alias="runId") + session_id: str = Field(..., alias="sessionId") + status: str + final_message: str = Field("", alias="finalMessage") + stop_reason: Optional[str] = Field(None, alias="stopReason") + error: Optional[str] = None + title: Optional[str] = None + tool_trace: List[ToolTraceEntryResponse] = Field( + default_factory=list, alias="toolTrace" + ) + usage: Dict[str, Any] = Field(default_factory=dict) + oauth_required: List[OAuthConsentRequiredResponse] = Field( + default_factory=list, alias="oauthRequired" + ) + started_at: str = Field("", alias="startedAt") + finished_at: str = Field("", alias="finishedAt") + + model_config = {"populate_by_name": True} + + @classmethod + def from_run_result(cls, result: RunResult) -> "RunNowResponse": + return cls( + run_id=result.run_id, + session_id=result.session_id, + status=result.status, + final_message=result.final_message, + stop_reason=result.stop_reason, + error=result.error, + title=result.title, + tool_trace=[ + ToolTraceEntryResponse( + tool_use_id=t.tool_use_id, + name=t.name, + input=t.input, + result_preview=t.result_preview, + is_error=t.is_error, + ) + for t in result.tool_trace + ], + usage=result.usage, + oauth_required=[ + OAuthConsentRequiredResponse( + provider_id=o.provider_id, + authorization_url=o.authorization_url, + ) + for o in result.oauth_required + ], + started_at=result.started_at, + finished_at=result.finished_at, + ) + + +class GrantStatusResponse(BaseModel): + enabled: bool + grant_id: Optional[str] = Field(None, alias="grantId") + created_at: Optional[int] = Field(None, alias="createdAt") + updated_at: Optional[int] = Field(None, alias="updatedAt") + expires_at: Optional[int] = Field(None, alias="expiresAt") + last_used_at: Optional[int] = Field(None, alias="lastUsedAt") + + model_config = {"populate_by_name": True} + + @classmethod + def from_grant(cls, grant: Optional[HeadlessGrant]) -> "GrantStatusResponse": + if grant is None: + return cls(enabled=False) + return cls( + enabled=True, + grant_id=grant.grant_id, + created_at=grant.created_at, + updated_at=grant.updated_at, + expires_at=grant.ttl, + last_used_at=grant.last_used_at, + ) + + +class GrantRevokeResponse(BaseModel): + revoked: bool + + +# ─── Routes ───────────────────────────────────────────────────────────────── + + +async def _resolve_grant(request: Request, user: User) -> HeadlessGrant: + """Create-on-enable: pin the live session's token, else use an existing grant. + + The BFF middleware attaches the caller's ``SessionRecord`` to + ``request.state.bff_session``; when present, the grant is created or + renewed from it (that session's ``created_at`` anchors the 30-day + login-recency window — see ``apis.shared.harness.grants``). Without a + session record (e.g. local SKIP_AUTH dev) an already-active grant still + works; having neither is a 409, not a 401 — the caller *is* + authenticated, they just have no credential the platform may act + headlessly with. + """ + grants = get_headless_grant_service() + session_record = getattr(request.state, "bff_session", None) + if session_record is not None: + return await grants.enable( + user_id=user.user_id, + username=session_record.username, + refresh_token=session_record.cognito_refresh_token, + token_issued_at=session_record.created_at, + ) + grant = await grants.get_active_grant(user.user_id) + if grant is None: + raise HTTPException( + status_code=status.HTTP_409_CONFLICT, + detail=( + "No headless grant exists for this account and the current " + "request carries no session to create one from." + ), + ) + return grant + + +@router.post("/now", response_model=RunNowResponse, response_model_by_alias=True) +async def run_now( + body: RunNowRequest, + request: Request, + user: User = Depends(require_scheduled_runs_user), +) -> RunNowResponse: + """Execute one agent turn as the caller through the headless harness. + + Synchronous from the caller's perspective: the response is the full + ``RunResult`` once the turn drains (bounded by the harness's 300s + budget, matching the chat proxy). The result also lands as a session in + the caller's conversation list, so a closed tab loses nothing. + """ + await _resolve_grant(request, user) + + try: + result = await run_agent_headless( + user_id=user.user_id, + prompt=body.prompt, + auth=CognitoRefreshBearerAuth(grants=get_headless_grant_service()), + title=body.title, + model_id=body.model_id, + rag_assistant_id=body.rag_assistant_id, + enabled_tools=body.enabled_tools, + agent_type=body.agent_type, + trigger="run_now", + ) + except HeadlessAuthError as exc: + # The grant exists but could not mint (Cognito refused — token + # expired or revoked upstream). 409, not 401: a 401 here would + # bounce the SPA through the login redirect even though the + # *session* is fine. + logger.warning("Run-now mint failed for user %s: %s", user.user_id, exc) + raise HTTPException( + status_code=status.HTTP_409_CONFLICT, + detail="Headless credential could not be minted; log in again to renew it.", + ) + + logger.info( + "Run-now %s for user %s finished status=%s session=%s", + result.run_id, + user.user_id, + result.status, + result.session_id, + ) + return RunNowResponse.from_run_result(result) + + +@router.get( + "/grant", response_model=GrantStatusResponse, response_model_by_alias=True +) +async def get_grant_status( + user: User = Depends(require_scheduled_runs_user), +) -> GrantStatusResponse: + """The caller's headless-grant status (never the token itself).""" + grant = await get_headless_grant_service().get_active_grant(user.user_id) + return GrantStatusResponse.from_grant(grant) + + +@router.delete("/grant", response_model=GrantRevokeResponse) +async def revoke_grant( + user: User = Depends(require_scheduled_runs_user), +) -> GrantRevokeResponse: + """Revoke the caller's headless grant (total revocation — the stored + credential is deleted in the same write).""" + revoked = await get_headless_grant_service().revoke(user.user_id) + return GrantRevokeResponse(revoked=revoked) diff --git a/backend/src/apis/shared/feature_flags.py b/backend/src/apis/shared/feature_flags.py index c7bbbec37..96cdd105b 100644 --- a/backend/src/apis/shared/feature_flags.py +++ b/backend/src/apis/shared/feature_flags.py @@ -1,9 +1,10 @@ """Process-level feature flags resolved from environment variables. -These gate optional product surfaces that ship in the codebase but stay -disabled for an environment until explicitly turned on — mirroring the -``FINE_TUNING_ENABLED`` pattern used in app-api. Each flag is read on every -call (not cached at import) so that: +These gate optional product surfaces per environment. Each flag documents +its own default: deferred features default off until explicitly turned on +(the ``FINE_TUNING_ENABLED`` pattern), while shipping features default on +with a kill switch (the ``KB_SYNC_ENABLED`` pattern). Each flag is read on +every call (not cached at import) so that: * import-time callers (conditional router mounting) and per-request callers observe the same value, and @@ -24,3 +25,22 @@ def skills_enabled() -> bool: unmounted / hidden, but all skills data and code remain intact. """ return os.environ.get("SKILLS_ENABLED", "false").lower() == "true" + + +def scheduled_runs_enabled() -> bool: + """Whether the scheduled-runs surface is enabled for this environment. + + Covers the headless "Run now" route and headless-grant lifecycle today, + and the schedule CRUD + dispatcher when Phase B lands. **Default ON + with a kill switch** (house style, mirroring ``CDK_KB_SYNC_ENABLED``): + unset or empty resolves to enabled; only the literal ``"false"`` + (case-insensitive) disables. The CDK side threads + ``config.scheduledRuns.enabled`` into this env var with the same + empty-string-safe ternary, so an unset GitHub Actions variable can + never silently turn the feature off. + + Note this flag gates *feature existence* per environment; *who* can use + it is the ``scheduled-runs`` RBAC capability + (``apis.shared.rbac.capabilities``) — two independent controls. + """ + return os.environ.get("SCHEDULED_RUNS_ENABLED", "").strip().lower() != "false" diff --git a/backend/src/apis/shared/rbac/capabilities.py b/backend/src/apis/shared/rbac/capabilities.py new file mode 100644 index 000000000..328eed0cf --- /dev/null +++ b/backend/src/apis/shared/rbac/capabilities.py @@ -0,0 +1,44 @@ +"""Feature-capability checks riding the AppRole grant system. + +A *capability* is a feature-level permission ("may this user use scheduled +runs?") rather than a resource-level one ("may this user call this tool?"). +Rather than adding a net-new allowlist table or a new grant axis for the +first capability, capability ids are granted through the mature **tools +grant axis**: an admin adds the capability id to a role's ``grantedTools`` +(e.g. a ``scheduled_runs_beta`` role granting ``scheduled-runs``), and this +module checks it via the same cached RBAC resolution path every tool check +uses (scheduled-agent-runs.md §6 — "reuses the mature RBAC resolution path, +no net-new allowlist table"). + +Consequences to be aware of: + +* A wildcard tools grant (``*`` — the seeded ``system_admin`` role) holds + every capability implicitly. Admins are in every beta by construction. +* Capability ids share a namespace with tool ids. They never collide with + a real tool in practice (no tool in the catalog is named like a feature), + and a granted capability id simply matches no tool at agent-build time — + but pick ids that read as features, not tools. +* GA for a capability = grant its id to the ``default`` role. One config + change, no redeploy (scheduled-agent-runs.md §6, "GA path"). + +If capabilities outgrow this (per-capability metadata, UI surfacing), the +RBAC gap ledger already names the real fix: extend the grant vocabulary +with a first-class axis (agentic-platform-primitives.md §1, RBAC row). +""" + +from __future__ import annotations + +from apis.shared.auth.models import User +from apis.shared.rbac.service import get_app_role_service + +#: Gates the headless-runs surface ("Run now" today; schedule CRUD in +#: Phase B). Granted to the beta cohort's role; GA = grant to ``default``. +SCHEDULED_RUNS_CAPABILITY = "scheduled-runs" + + +async def user_has_capability(user: User, capability_id: str) -> bool: + """True iff ``user`` resolves the capability through their AppRoles. + + Wildcard tool grants satisfy every capability (see module docstring). + """ + return await get_app_role_service().can_access_tool(user, capability_id) diff --git a/backend/tests/apis/app_api/test_runs_routes.py b/backend/tests/apis/app_api/test_runs_routes.py new file mode 100644 index 000000000..3a2c71e99 --- /dev/null +++ b/backend/tests/apis/app_api/test_runs_routes.py @@ -0,0 +1,345 @@ +"""Route tests for the headless "Run now" surface (`/runs/*`). + +Pins the three-layer gate (cookie auth → SCHEDULED_RUNS_ENABLED kill switch +→ `scheduled-runs` RBAC capability), the create-on-enable grant flow, and +the RunResult → camelCase response mapping. +""" + +from __future__ import annotations + +import time +from typing import Optional + +import pytest +from fastapi import FastAPI, Request +from fastapi.testclient import TestClient + +from apis.shared.auth.dependencies import get_current_user_from_session +from apis.shared.auth.models import User +from apis.shared.harness.auth import HeadlessAuthError +from apis.shared.harness.grants import HeadlessGrant +from apis.shared.harness.models import RunResult, ToolTraceEntry + +from apis.app_api.runs import routes as runs_routes + +NOW = int(time.time()) + + +def _user() -> User: + return User( + user_id="user-1", + email="user@example.com", + name="User", + roles=["default"], + raw_token="tok", + ) + + +def _grant(user_id: str = "user-1") -> HeadlessGrant: + return HeadlessGrant( + grant_id="hlg-abc", + user_id=user_id, + username="user1", + cognito_refresh_token="rt-stored", + status="active", + created_at=NOW - 100, + updated_at=NOW - 100, + token_issued_at=NOW - 100, + ttl=NOW + 1000, + last_used_at=NOW - 50, + ) + + +class FakeGrantService: + def __init__(self, grant: Optional[HeadlessGrant] = None): + self.grant = grant + self.enable_calls: list[dict] = [] + self.revoke_calls: list[str] = [] + + async def get_active_grant(self, user_id: str): + return self.grant + + async def enable(self, *, user_id, username, refresh_token, token_issued_at=None): + self.enable_calls.append( + { + "user_id": user_id, + "username": username, + "refresh_token": refresh_token, + "token_issued_at": token_issued_at, + } + ) + self.grant = _grant(user_id) + return self.grant + + async def revoke(self, user_id: str) -> bool: + self.revoke_calls.append(user_id) + revoked = self.grant is not None + self.grant = None + return revoked + + +class FakeSessionRecord: + """Just the SessionRecord fields the route reads.""" + + username = "user1" + cognito_refresh_token = "rt-live" + created_at = NOW - 3600 + + +def _completed_result() -> RunResult: + return RunResult( + run_id="run-1", + session_id="headless-1", + user_id="user-1", + status="completed", + final_message="pong", + stop_reason="end_turn", + title="T", + tool_trace=[ + ToolTraceEntry( + tool_use_id="t1", + name="search_classes", + input={"subject": "COMM"}, + result_preview="ok", + ) + ], + usage={"usage": {"totalTokens": 6}}, + started_at="2026-07-05T00:00:00Z", + finished_at="2026-07-05T00:00:10Z", + ) + + +def _make_client( + monkeypatch: pytest.MonkeyPatch, + *, + authed: bool = True, + capability: bool = True, + flag: Optional[str] = None, + grants: Optional[FakeGrantService] = None, + with_session_record: bool = False, + run_result: Optional[RunResult] = None, + run_error: Optional[Exception] = None, +) -> tuple[TestClient, FakeGrantService, list[dict]]: + monkeypatch.delenv("SKIP_AUTH", raising=False) + if flag is None: + monkeypatch.delenv("SCHEDULED_RUNS_ENABLED", raising=False) + else: + monkeypatch.setenv("SCHEDULED_RUNS_ENABLED", flag) + + async def fake_capability(user, capability_id): + assert capability_id == "scheduled-runs" + return capability + + monkeypatch.setattr(runs_routes, "user_has_capability", fake_capability) + + grants = grants or FakeGrantService() + monkeypatch.setattr(runs_routes, "get_headless_grant_service", lambda: grants) + + run_calls: list[dict] = [] + + async def fake_run(**kwargs): + run_calls.append(kwargs) + if run_error is not None: + raise run_error + return run_result or _completed_result() + + monkeypatch.setattr(runs_routes, "run_agent_headless", fake_run) + + app = FastAPI() + + if with_session_record: + @app.middleware("http") + async def attach_session(request: Request, call_next): + request.state.bff_session = FakeSessionRecord() + return await call_next(request) + + app.include_router(runs_routes.router) + if authed: + app.dependency_overrides[get_current_user_from_session] = _user + client = TestClient(app, raise_server_exceptions=False) + return client, grants, run_calls + + +# --------------------------------------------------------------------------- +# scheduled_runs_enabled() helper — default ON with a kill switch +# --------------------------------------------------------------------------- + + +class TestScheduledRunsFlag: + @pytest.mark.parametrize( + "value, expected", + [ + (None, True), # unset → default on + ("", True), # empty workflow var → default on + ("true", True), + ("false", False), + ("FALSE", False), + ("0", True), # only the literal "false" disables + ], + ) + def test_parses_env_value(self, monkeypatch, value, expected): + from apis.shared.feature_flags import scheduled_runs_enabled + + if value is None: + monkeypatch.delenv("SCHEDULED_RUNS_ENABLED", raising=False) + else: + monkeypatch.setenv("SCHEDULED_RUNS_ENABLED", value) + assert scheduled_runs_enabled() is expected + + +# --------------------------------------------------------------------------- +# Gating +# --------------------------------------------------------------------------- + + +class TestGating: + def test_unauthenticated_request_is_401(self, monkeypatch): + client, _, _ = _make_client(monkeypatch, authed=False) + assert client.post("/runs/now", json={"prompt": "hi"}).status_code == 401 + + def test_kill_switch_off_hides_the_surface_as_404(self, monkeypatch): + client, _, _ = _make_client(monkeypatch, flag="false") + assert client.post("/runs/now", json={"prompt": "hi"}).status_code == 404 + assert client.get("/runs/grant").status_code == 404 + assert client.delete("/runs/grant").status_code == 404 + + def test_flag_defaults_on_when_unset(self, monkeypatch): + client, _, _ = _make_client(monkeypatch, with_session_record=True) + assert client.post("/runs/now", json={"prompt": "hi"}).status_code == 200 + + def test_empty_flag_value_stays_on(self, monkeypatch): + # `${{ vars.* }}` renders "" when unset — must resolve to the default. + client, _, _ = _make_client(monkeypatch, flag="", with_session_record=True) + assert client.post("/runs/now", json={"prompt": "hi"}).status_code == 200 + + def test_missing_capability_is_403(self, monkeypatch): + client, _, _ = _make_client(monkeypatch, capability=False) + response = client.post("/runs/now", json={"prompt": "hi"}) + assert response.status_code == 403 + assert client.get("/runs/grant").status_code == 403 + + +# --------------------------------------------------------------------------- +# POST /runs/now +# --------------------------------------------------------------------------- + + +class TestRunNow: + def test_happy_path_maps_run_result_to_camel_case(self, monkeypatch): + client, grants, run_calls = _make_client( + monkeypatch, with_session_record=True + ) + + response = client.post( + "/runs/now", + json={ + "prompt": "ping", + "title": "My Briefing", + "enabledTools": ["class_search"], + "agentType": "chat", + }, + ) + + assert response.status_code == 200 + body = response.json() + assert body["runId"] == "run-1" + assert body["sessionId"] == "headless-1" + assert body["status"] == "completed" + assert body["finalMessage"] == "pong" + assert body["stopReason"] == "end_turn" + assert body["toolTrace"] == [ + { + "toolUseId": "t1", + "name": "search_classes", + "input": {"subject": "COMM"}, + "resultPreview": "ok", + "isError": False, + } + ] + assert body["usage"]["usage"]["totalTokens"] == 6 + + (call,) = run_calls + assert call["user_id"] == "user-1" + assert call["prompt"] == "ping" + assert call["title"] == "My Briefing" + assert call["enabled_tools"] == ["class_search"] + assert call["agent_type"] == "chat" + assert call["trigger"] == "run_now" + + def test_create_on_enable_pins_the_live_session_token(self, monkeypatch): + client, grants, _ = _make_client(monkeypatch, with_session_record=True) + + client.post("/runs/now", json={"prompt": "ping"}) + + (enable,) = grants.enable_calls + assert enable["user_id"] == "user-1" + assert enable["username"] == "user1" + assert enable["refresh_token"] == "rt-live" + # The session's login instant anchors the 30-day recency window. + assert enable["token_issued_at"] == NOW - 3600 + + def test_existing_grant_works_without_a_session_record(self, monkeypatch): + grants = FakeGrantService(grant=_grant()) + client, grants, _ = _make_client(monkeypatch, grants=grants) + + assert client.post("/runs/now", json={"prompt": "ping"}).status_code == 200 + assert grants.enable_calls == [] # no session to re-pin from + + def test_no_grant_and_no_session_is_409(self, monkeypatch): + client, _, run_calls = _make_client(monkeypatch) + + response = client.post("/runs/now", json={"prompt": "ping"}) + + assert response.status_code == 409 + assert run_calls == [] # never reached the harness + + def test_mint_failure_is_409_not_401(self, monkeypatch): + # 401 would bounce the SPA through the login redirect; the *session* + # is fine — only the headless credential is dead. + client, _, _ = _make_client( + monkeypatch, + with_session_record=True, + run_error=HeadlessAuthError("cognito refused"), + ) + + assert client.post("/runs/now", json={"prompt": "ping"}).status_code == 409 + + def test_empty_prompt_is_422(self, monkeypatch): + client, _, _ = _make_client(monkeypatch, with_session_record=True) + assert client.post("/runs/now", json={"prompt": ""}).status_code == 422 + + +# --------------------------------------------------------------------------- +# Grant lifecycle routes +# --------------------------------------------------------------------------- + + +class TestGrantRoutes: + def test_grant_status_when_enabled(self, monkeypatch): + client, _, _ = _make_client(monkeypatch, grants=FakeGrantService(_grant())) + + body = client.get("/runs/grant").json() + + assert body["enabled"] is True + assert body["grantId"] == "hlg-abc" + assert body["expiresAt"] == NOW + 1000 + # The stored credential itself must never surface. + assert "rt-stored" not in str(body) + + def test_grant_status_when_absent(self, monkeypatch): + client, _, _ = _make_client(monkeypatch) + assert client.get("/runs/grant").json() == { + "enabled": False, + "grantId": None, + "createdAt": None, + "updatedAt": None, + "expiresAt": None, + "lastUsedAt": None, + } + + def test_revoke_grant(self, monkeypatch): + client, grants, _ = _make_client(monkeypatch, grants=FakeGrantService(_grant())) + + assert client.delete("/runs/grant").json() == {"revoked": True} + assert grants.revoke_calls == ["user-1"] + assert client.delete("/runs/grant").json() == {"revoked": False} From 664e98719262c780eec13a84ed4cec75c3a98b3d Mon Sep 17 00:00:00 2001 From: Phil Merrell Date: Sun, 5 Jul 2026 18:11:27 -0600 Subject: [PATCH 4/4] feat(infra): SCHEDULED_RUNS_ENABLED flag + HeadlessGrantUserIndex GSI MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Thread the scheduled-runs kill switch through CDK: scheduledRuns.enabled in config.ts (the CDK_KB_SYNC_ENABLED empty-string-safe ternary, copied exactly) -> SCHEDULED_RUNS_ENABLED on the app-api container -> CDK_SCHEDULED_RUNS_ENABLED forwarded by platform.yml. Default ON with a kill switch; nightly inherits the default. Add the sparse HeadlessGrantUserIndex GSI (grant_user_id / created_at) to the BFF sessions table backing apis/shared/harness/grants.py — only HEADLESS-GRANT# items carry the partition attribute, so session rows never project into it. App-api's existing table grant already covers index/*, so no IAM change. Tests mirror the kbSync flag matrix and pin the GSI shape. Co-Authored-By: Claude Opus 4.8 --- .github/workflows/platform.yml | 6 +++ infrastructure/lib/config.ts | 28 +++++++++++++ .../constructs/app-api/app-api-environment.ts | 4 ++ .../constructs/data/auth-tables-construct.ts | 12 ++++++ infrastructure/test/config.test.ts | 39 +++++++++++++++++++ infrastructure/test/helpers/mock-config.ts | 3 ++ infrastructure/test/tables-detailed.test.ts | 19 +++++++++ 7 files changed, 111 insertions(+) diff --git a/.github/workflows/platform.yml b/.github/workflows/platform.yml index c91478c32..335b71352 100644 --- a/.github/workflows/platform.yml +++ b/.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 }} diff --git a/infrastructure/lib/config.ts b/infrastructure/lib/config.ts index 7cf64be52..c081f3ce7 100644 --- a/infrastructure/lib/config.ts +++ b/infrastructure/lib/config.ts @@ -49,6 +49,7 @@ export interface AppConfig { inferenceApi: InferenceApiConfig; ragIngestion: RagIngestionConfig; kbSync: KbSyncConfig; + scheduledRuns: ScheduledRunsConfig; fineTuning: FineTuningConfig; artifacts: ArtifactsConfig; mcpSandbox: McpSandboxConfig; @@ -140,6 +141,22 @@ export interface KbSyncConfig { enabled: boolean; } +/** + * Scheduled runs — headless agent runs as a user (the Harness primitive, + * docs/specs/scheduled-agent-runs.md). + * + * `enabled` is the global kill switch: it sets the SCHEDULED_RUNS_ENABLED + * env var on app-api (gating the "Run now" + headless-grant routes), and + * will gate the Phase-B EventBridge dispatcher rule when the scheduler + * lands. Default ON with a kill switch — the feature runs unless it's + * explicitly turned off with CDK_SCHEDULED_RUNS_ENABLED=false (or a + * `scheduledRuns.enabled: false` cdk.json context). *Who* can use the + * surface is governed separately by the `scheduled-runs` RBAC capability. + */ +export interface ScheduledRunsConfig { + enabled: boolean; +} + export interface FineTuningConfig { additionalCorsOrigins?: string; // Extra CORS origins to append (comma-separated) } @@ -276,6 +293,17 @@ export function loadConfig(scope: cdk.App): AppConfig { ? process.env.CDK_KB_SYNC_ENABLED !== 'false' : scope.node.tryGetContext('kbSync')?.enabled ?? true, }, + scheduledRuns: { + // Default ON with a kill switch: enabled unless explicitly disabled. + // The workflow forwards `${{ vars.CDK_SCHEDULED_RUNS_ENABLED }}`, + // which is an EMPTY STRING when the variable is unset — so treat + // empty/unset as "use the default (on)" and only the literal "false" + // as the off switch. A `scheduledRuns.enabled` cdk.json context can + // also force it off. (Same ternary as kbSync above — keep in sync.) + enabled: process.env.CDK_SCHEDULED_RUNS_ENABLED + ? process.env.CDK_SCHEDULED_RUNS_ENABLED !== 'false' + : scope.node.tryGetContext('scheduledRuns')?.enabled ?? true, + }, fineTuning: { additionalCorsOrigins: process.env.CDK_FINE_TUNING_CORS_ORIGINS || scope.node.tryGetContext('fineTuning')?.additionalCorsOrigins, }, diff --git a/infrastructure/lib/constructs/app-api/app-api-environment.ts b/infrastructure/lib/constructs/app-api/app-api-environment.ts index 8de73a3ee..22333354d 100644 --- a/infrastructure/lib/constructs/app-api/app-api-environment.ts +++ b/infrastructure/lib/constructs/app-api/app-api-environment.ts @@ -273,6 +273,10 @@ export function buildAppApiEnvironment( ? `https://${config.domainName}/` : 'http://localhost:4200/', INFERENCE_API_URL: params.inferenceApiRuntimeEndpointUrl, + // Kill switch for the headless "Run now" + headless-grant routes + // (scheduled-runs PR-1). Cohort access is RBAC (`scheduled-runs` + // capability); this only gates feature existence per environment. + SCHEDULED_RUNS_ENABLED: config.scheduledRuns.enabled ? 'true' : 'false', VOICE_TICKET_REPLAY_TABLE_NAME: params.voiceTicketReplayTableName, VOICE_TICKET_SIGNING_SECRET_ARN: params.voiceTicketSigningSecretArn, }; diff --git a/infrastructure/lib/constructs/data/auth-tables-construct.ts b/infrastructure/lib/constructs/data/auth-tables-construct.ts index d8d84bf6e..6fa55f60f 100644 --- a/infrastructure/lib/constructs/data/auth-tables-construct.ts +++ b/infrastructure/lib/constructs/data/auth-tables-construct.ts @@ -54,6 +54,18 @@ export class AuthTablesConstruct extends Construct { encryption: dynamodb.TableEncryption.AWS_MANAGED, }); + // HeadlessGrantUserIndex — sparse reverse lookup of headless-run grants + // by owner. Grant items (PK=HEADLESS-GRANT#{id}) are the only rows that + // carry `grant_user_id`, so ordinary session rows never project here. + // Backs apis/shared/harness/grants.py (scheduled-runs PR-1): the + // per-owner grant query that replaced the spike's full-table Scan. + this.bffSessionsTable.addGlobalSecondaryIndex({ + indexName: 'HeadlessGrantUserIndex', + partitionKey: { name: 'grant_user_id', type: dynamodb.AttributeType.STRING }, + sortKey: { name: 'created_at', type: dynamodb.AttributeType.NUMBER }, + projectionType: dynamodb.ProjectionType.ALL, + }); + // Users Table - User profiles synced from JWT diff --git a/infrastructure/test/config.test.ts b/infrastructure/test/config.test.ts index cc34d30b3..cd0da833a 100644 --- a/infrastructure/test/config.test.ts +++ b/infrastructure/test/config.test.ts @@ -345,6 +345,45 @@ describe('RAG Ingestion Configuration', () => { }); }); + // ============================================================ + // Scheduled Runs feature flag — default ON with a kill switch + // (same ternary as kbSync; empty workflow var must not disable) + // ============================================================ + + describe('Scheduled Runs feature flag', () => { + test('defaults to enabled when CDK_SCHEDULED_RUNS_ENABLED is unset', () => { + delete process.env.CDK_SCHEDULED_RUNS_ENABLED; + + expect(loadConfig(app).scheduledRuns.enabled).toBe(true); + }); + + test('treats empty string (unset GitHub Actions variable) as enabled', () => { + // `${{ vars.CDK_SCHEDULED_RUNS_ENABLED }}` renders to "" when unset. + process.env.CDK_SCHEDULED_RUNS_ENABLED = ''; + + expect(loadConfig(app).scheduledRuns.enabled).toBe(true); + }); + + test('CDK_SCHEDULED_RUNS_ENABLED="false" is the kill switch', () => { + process.env.CDK_SCHEDULED_RUNS_ENABLED = 'false'; + + expect(loadConfig(app).scheduledRuns.enabled).toBe(false); + }); + + test('CDK_SCHEDULED_RUNS_ENABLED="true" stays enabled', () => { + process.env.CDK_SCHEDULED_RUNS_ENABLED = 'true'; + + expect(loadConfig(app).scheduledRuns.enabled).toBe(true); + }); + + test('cdk.json context scheduledRuns.enabled=false disables when env is unset', () => { + delete process.env.CDK_SCHEDULED_RUNS_ENABLED; + app.node.setContext('scheduledRuns', { enabled: false }); + + expect(loadConfig(app).scheduledRuns.enabled).toBe(false); + }); + }); + // ============================================================ // Configuration Validation Tests // ============================================================ diff --git a/infrastructure/test/helpers/mock-config.ts b/infrastructure/test/helpers/mock-config.ts index 9d58a570d..3529db781 100644 --- a/infrastructure/test/helpers/mock-config.ts +++ b/infrastructure/test/helpers/mock-config.ts @@ -48,6 +48,9 @@ export function createMockConfig(overrides: Partial = {}): AppConfig kbSync: { enabled: false, }, + scheduledRuns: { + enabled: false, + }, fineTuning: {}, artifacts: { retentionDays: 90, diff --git a/infrastructure/test/tables-detailed.test.ts b/infrastructure/test/tables-detailed.test.ts index 9f6bc138a..2e56fdcbc 100644 --- a/infrastructure/test/tables-detailed.test.ts +++ b/infrastructure/test/tables-detailed.test.ts @@ -55,6 +55,25 @@ describe('AuthTablesConstruct — detailed', () => { }); }); + it('BFFSessions table has the sparse HeadlessGrantUserIndex GSI', () => { + // Backs apis/shared/harness/grants.py — per-owner headless-grant lookup. + // Sparse: only HEADLESS-GRANT# items carry grant_user_id, so session + // rows never project into it. + t.hasResourceProperties('AWS::DynamoDB::Table', { + TableName: 'test-project-bff-sessions', + GlobalSecondaryIndexes: Match.arrayWith([ + Match.objectLike({ + IndexName: 'HeadlessGrantUserIndex', + KeySchema: [ + { AttributeName: 'grant_user_id', KeyType: 'HASH' }, + { AttributeName: 'created_at', KeyType: 'RANGE' }, + ], + Projection: { ProjectionType: 'ALL' }, + }), + ]), + }); + }); + it('Users table has 4 GSIs', () => { t.hasResourceProperties('AWS::DynamoDB::Table', { TableName: 'test-project-users',