diff --git a/agent/src/observability.py b/agent/src/observability.py index 0ca2ee29..526c437d 100644 --- a/agent/src/observability.py +++ b/agent/src/observability.py @@ -64,24 +64,54 @@ def current_otel_trace_id() -> str | None: bundle) so operators can join the task to its CloudWatch/X-Ray trace. Returns ``None`` when there is no recording span (e.g. tracing disabled locally) or the context is invalid, so callers can treat it as a graceful-missing field. + + Never raises: a broken/misconfigured tracer degrades to ``None`` rather than + propagating. Callers read this inside DDB-write try-blocks (progress_writer), + where a raised trace error would otherwise be misclassified as a DDB failure + and trip the shared progress circuit breaker. """ - span = trace.get_current_span() - ctx = span.get_span_context() - if not ctx.is_valid: + try: + span = trace.get_current_span() + ctx = span.get_span_context() + if not ctx.is_valid: + return None + # format_trace_id renders the 128-bit id as zero-padded 32-char hex — the + # OTEL format, so it joins directly in CloudWatch Transaction Search. Note + # the X-Ray console renders trace ids as ``1-{8hex}-{24hex}``; to look this + # up there, transform to that form (the timestamp is the first 8 hex chars). + return trace.format_trace_id(ctx.trace_id) + except Exception: + # nosemgrep: py-silent-success-masking -- trace id is a graceful-missing + # correlation field; a tracer fault must not fail the caller's write path. return None - # format_trace_id renders the 128-bit id as zero-padded 32-char hex — the - # OTEL format, so it joins directly in CloudWatch Transaction Search. Note - # the X-Ray console renders trace ids as ``1-{8hex}-{24hex}``; to look this - # up there, transform to that form (the timestamp is the first 8 hex chars). - return trace.format_trace_id(ctx.trace_id) -def set_session_id(session_id: str) -> None: - """Propagate *session_id* via OTEL baggage for AgentCore session correlation. +def propagate_correlation_context( + session_id: str, + user_id: str = "", + # ``user_id`` uses ""-means-absent (Cognito sub, mirrors AgentConfig.user_id + # which is never None); ``repo`` uses None-means-absent (mirrors the optional + # TaskRecord.repo). Both conventions flow from upstream config types; the + # ``if x:`` guards below flatten either to "don't set the baggage key". + repo: str | None = None, +) -> None: + """Propagate the correlation envelope via OTEL baggage. + + *session_id* correlates custom spans to the AgentCore session; *user_id* + and *repo* (#245) carry the platform identity and target repo so baggage + survives across pipeline phases on the task thread. Empty/None fields are + not set — so this runs (and is useful) even when *session_id* is empty but + the identity is known. *repo* is None for repo-less workflows (#248 Phase 3). The attached context is intentionally not detached: the background thread runs a single task then exits, so the context is garbage-collected with the thread. """ - ctx = baggage.set_baggage("session.id", session_id) + ctx = context.get_current() + if session_id: + ctx = baggage.set_baggage("session.id", session_id, context=ctx) + if user_id: + ctx = baggage.set_baggage("user.id", user_id, context=ctx) + if repo: + ctx = baggage.set_baggage("repo.url", repo, context=ctx) context.attach(ctx) # token not stored — thread-scoped lifetime diff --git a/agent/src/pipeline.py b/agent/src/pipeline.py index 95eaaff8..a6bb7857 100644 --- a/agent/src/pipeline.py +++ b/agent/src/pipeline.py @@ -679,13 +679,18 @@ def run_task( "repo.url": config.repo_url, "issue.number": config.issue_number, "agent.model": config.anthropic_model, + # Correlation envelope (#245): user.id joins agent spans to + # orchestrator logs by the platform identity, not just task/repo. + **({"user.id": config.user_id} if config.user_id else {}), }, ) as root_span: task_state.write_running(config.task_id) task_state.write_heartbeat(config.task_id) agent_result: AgentResult | None = None - progress = _ProgressWriter(config.task_id, trace=trace) + progress = _ProgressWriter( + config.task_id, trace=trace, user_id=config.user_id, repo=config.repo_url + ) # #251: clear any blocker latched by a prior task. The agent container # is one-task-per-process today, but the FastAPI server thread-pool can # in principle dispatch a second run_task in the same process — reset diff --git a/agent/src/progress_writer.py b/agent/src/progress_writer.py index bd2c08e3..f7dbea69 100644 --- a/agent/src/progress_writer.py +++ b/agent/src/progress_writer.py @@ -38,6 +38,8 @@ from decimal import Decimal from typing import Literal +from observability import current_otel_trace_id + # Preview field cap defaults (design §10.1): # - 200 chars for normal tasks — small DDB rows, cheap watch-stream bytes. # - 4096 chars (4 KB) for ``--trace`` opt-in tasks — full enough to @@ -373,8 +375,19 @@ class _ProgressWriter: _MAX_FAILURES = 3 - def __init__(self, task_id: str, trace: bool = False) -> None: + def __init__( + self, + task_id: str, + trace: bool = False, + user_id: str = "", + repo: str | None = None, + ) -> None: self._task_id = task_id + # Correlation envelope (#245): stamped on every event so the agent-side + # event stream joins to orchestrator logs and the X-Ray trace. ``repo`` + # is None for repo-less workflows (#248 Phase 3). + self._user_id = user_id + self._repo = repo self._table_name = os.environ.get("TASK_EVENTS_TABLE_NAME") self._table = None # Per-instance preview cap — design §10.1. ``trace=True`` raises @@ -467,6 +480,11 @@ def _put_event(self, event_type: str, metadata: dict) -> None: return now = datetime.now(UTC) + # Correlation envelope (#245): trace_id is read per-event from the + # active span (it may be None early, before the root span opens); + # user_id/repo are stamped when known. All are omitted when + # empty/None so the item shape stays stable for consumers. + trace_id = current_otel_trace_id() item = { "task_id": self._task_id, "event_id": _generate_ulid(), @@ -478,6 +496,12 @@ def _put_event(self, event_type: str, metadata: dict) -> None: "timestamp": now.isoformat(), "ttl": int(now.timestamp()) + _TTL_SECONDS, } + if self._user_id: + item["user_id"] = self._user_id + if self._repo: + item["repo"] = self._repo + if trace_id: + item["trace_id"] = trace_id self._table.put_item(Item=item) # Success: reset the shared failure counter. We do NOT flip diff --git a/agent/src/runner.py b/agent/src/runner.py index bed2741f..94902197 100644 --- a/agent/src/runner.py +++ b/agent/src/runner.py @@ -483,7 +483,12 @@ def _on_stderr(line: str) -> None: # invocations we fall back to a fresh writer with no accumulator. if trajectory is None: trajectory = _TrajectoryWriter(config.task_id or "unknown") - progress = _ProgressWriter(config.task_id or "unknown", trace=config.trace) + progress = _ProgressWriter( + config.task_id or "unknown", + trace=config.trace, + user_id=config.user_id, + repo=config.repo_url, + ) # Map tool_use_id → tool_name so we can label ToolResultBlocks that arrive # in UserMessages (ToolResultBlock carries only the id, not the name). diff --git a/agent/src/server.py b/agent/src/server.py index 2916faa2..a7ae76ce 100644 --- a/agent/src/server.py +++ b/agent/src/server.py @@ -26,7 +26,7 @@ import task_state from config import resolve_github_token from models import TaskResult -from observability import set_session_id +from observability import propagate_correlation_context from pipeline import run_task # --- _debug_cw / _warn_cw failure counter ------------------------------- @@ -452,10 +452,12 @@ def _run_task_background( hb_thread.start() try: - # Propagate session ID into this thread's OTEL context so spans - # are correlated with the AgentCore session in CloudWatch. - if session_id: - set_session_id(session_id) + # Propagate the correlation envelope into this thread's OTEL context + # so spans are correlated with the AgentCore session and the platform + # identity in CloudWatch (#245). Runs whenever any field is present — + # session_id may be empty while user_id/repo are known. + if session_id or user_id or repo_url: + propagate_correlation_context(session_id, user_id=user_id, repo=repo_url or None) run_task( repo_url=repo_url, diff --git a/agent/tests/test_observability.py b/agent/tests/test_observability.py index ad5cd035..d07b8d49 100644 --- a/agent/tests/test_observability.py +++ b/agent/tests/test_observability.py @@ -44,3 +44,46 @@ def test_returns_none_for_invalid_context(self): span.get_span_context.return_value = ctx with patch.object(observability.trace, "get_current_span", return_value=span): assert observability.current_otel_trace_id() is None + + def test_degrades_to_none_when_tracer_raises(self): + # A broken/misconfigured tracer must not propagate: callers read this + # inside DDB-write try-blocks, where a raise would be misclassified as a + # DDB failure and trip the shared progress circuit breaker (#245 review). + with patch.object( + observability.trace, "get_current_span", side_effect=RuntimeError("tracer boom") + ): + assert observability.current_otel_trace_id() is None + + +class TestPropagateCorrelationContext: + """``propagate_correlation_context`` propagates the correlation envelope + (#245) via OTEL baggage, setting only the fields that are present.""" + + def test_sets_all_envelope_fields_when_present(self): + with ( + patch.object(observability, "baggage") as bag, + patch.object(observability, "context") as ctx, + ): + bag.set_baggage.return_value = "CTX" + observability.propagate_correlation_context("sess-1", user_id="user-1", repo="org/repo") + # Baggage-key ordering is not part of the contract — compare as a set. + keys = {c.args[0] for c in bag.set_baggage.call_args_list} + assert keys == {"session.id", "user.id", "repo.url"} + ctx.attach.assert_called_once() + + def test_omits_absent_fields(self): + # Empty user_id and None repo → only session.id is set on the baggage. + with patch.object(observability, "baggage") as bag, patch.object(observability, "context"): + bag.set_baggage.return_value = "CTX" + observability.propagate_correlation_context("sess-1") + keys = {c.args[0] for c in bag.set_baggage.call_args_list} + assert keys == {"session.id"} + + def test_empty_session_id_is_not_stamped(self): + # Reachable via server.py's widened trigger (user_id known, no session): + # an empty session_id must not write an empty-string session.id baggage. + with patch.object(observability, "baggage") as bag, patch.object(observability, "context"): + bag.set_baggage.return_value = "CTX" + observability.propagate_correlation_context("", user_id="user-1") + keys = {c.args[0] for c in bag.set_baggage.call_args_list} + assert keys == {"user.id"} diff --git a/agent/tests/test_progress_writer.py b/agent/tests/test_progress_writer.py index b0e76828..a837e7d2 100644 --- a/agent/tests/test_progress_writer.py +++ b/agent/tests/test_progress_writer.py @@ -267,6 +267,59 @@ def test_ttl_is_90_days_from_now(self, writer, mock_table): assert before + ttl_90_days <= item["ttl"] <= after + ttl_90_days + 1 +# --------------------------------------------------------------------------- +# _ProgressWriter — correlation envelope (#245) +# --------------------------------------------------------------------------- + + +class TestCorrelationEnvelope: + """Every emitted event stamps {user_id, repo, trace_id} when known, so the + agent-side event stream joins to orchestrator logs and the X-Ray trace.""" + + def _writer(self, monkeypatch, **kwargs): + monkeypatch.setenv("TASK_EVENTS_TABLE_NAME", "events-table") + writer = _ProgressWriter("task-42", **kwargs) + writer._table = MagicMock() + return writer + + def test_stamps_user_repo_trace_when_present(self, monkeypatch): + writer = self._writer(monkeypatch, user_id="user-1", repo="org/repo") + with patch("progress_writer.current_otel_trace_id", return_value="a" * 32): + writer.write_agent_milestone("m", "") + item = writer._table.put_item.call_args[1]["Item"] + assert item["user_id"] == "user-1" + assert item["repo"] == "org/repo" + assert item["trace_id"] == "a" * 32 + + def test_repo_less_omits_repo(self, monkeypatch): + # repo-less workflow (#248 Phase 3): repo is None → key omitted, not null. + writer = self._writer(monkeypatch, user_id="user-1", repo=None) + with patch("progress_writer.current_otel_trace_id", return_value="b" * 32): + writer.write_agent_milestone("m", "") + item = writer._table.put_item.call_args[1]["Item"] + assert item["user_id"] == "user-1" + assert "repo" not in item + + def test_missing_trace_id_omits_key(self, monkeypatch): + # No recording span (tracing disabled) → trace_id key omitted entirely. + writer = self._writer(monkeypatch, user_id="user-1", repo="org/repo") + with patch("progress_writer.current_otel_trace_id", return_value=None): + writer.write_agent_milestone("m", "") + item = writer._table.put_item.call_args[1]["Item"] + assert "trace_id" not in item + + def test_defaults_omit_all_correlation_keys(self, monkeypatch): + # Back-compat: a writer built without the envelope (empty user_id) stamps + # none of the optional keys, keeping the historical item shape. + writer = self._writer(monkeypatch) + with patch("progress_writer.current_otel_trace_id", return_value=None): + writer.write_agent_milestone("m", "") + item = writer._table.put_item.call_args[1]["Item"] + assert "user_id" not in item + assert "repo" not in item + assert "trace_id" not in item + + # --------------------------------------------------------------------------- # _ProgressWriter — --trace preview cap (design §10.1) # --------------------------------------------------------------------------- diff --git a/agent/tests/test_server.py b/agent/tests/test_server.py index 144f9514..a839cd0d 100644 --- a/agent/tests/test_server.py +++ b/agent/tests/test_server.py @@ -288,6 +288,41 @@ def fake_run_task(**_kwargs): assert heartbeat_calls[0] == "t-heartbeat" +def test_run_task_background_propagates_correlation_envelope(monkeypatch): + """The background task thread propagates {session_id, user_id, repo} into + OTEL baggage via propagate_correlation_context (#245). + + Regression guard for the widened trigger: correlation must propagate even + when session_id is empty but user_id/repo are known — the branch the whole + envelope-in-baggage feature depends on. + """ + calls: list[dict] = [] + monkeypatch.setattr( + server, + "propagate_correlation_context", + lambda session_id, **kw: calls.append({"session_id": session_id, **kw}), + ) + monkeypatch.setattr(server, "run_task", lambda **_kwargs: None) + monkeypatch.setattr(server.task_state, "write_heartbeat", lambda *a, **kw: None) + monkeypatch.setattr(server.task_state, "write_terminal", lambda *a, **kw: None) + + # No session_id, but user_id + repo_url known → propagation must still run. + server._run_task_background( + task_id="t-corr", + repo_url="o/r", + task_description="x", + issue_number="", + github_token="", + anthropic_model="", + max_turns=10, + max_budget_usd=None, + aws_region="us-east-1", + user_id="user-1", + ) + + assert calls == [{"session_id": "", "user_id": "user-1", "repo": "o/r"}] + + def test_validate_required_params_pr_workflows_require_pr_number(): """PR-iteration and PR-review workflows need a pr_number regardless.""" missing = server._validate_required_params( diff --git a/cdk/src/handlers/get-task-events.ts b/cdk/src/handlers/get-task-events.ts index f9b2a58a..12c01ce4 100644 --- a/cdk/src/handlers/get-task-events.ts +++ b/cdk/src/handlers/get-task-events.ts @@ -219,12 +219,18 @@ export async function handler(event: APIGatewayProxyEvent): Promise ({ event_id: e.event_id, event_type: e.event_type, timestamp: e.timestamp, metadata: e.metadata ?? {}, + ...(e.user_id && { user_id: e.user_id }), + ...(e.repo && { repo: e.repo }), + ...(e.trace_id && { trace_id: e.trace_id }), })); // For descending scans we intentionally suppress ``next_token``. DDB's diff --git a/cdk/src/handlers/get-task-replay.ts b/cdk/src/handlers/get-task-replay.ts index fdac52b8..705dcd17 100644 --- a/cdk/src/handlers/get-task-replay.ts +++ b/cdk/src/handlers/get-task-replay.ts @@ -197,6 +197,10 @@ export function assembleBundle( event_type: e.event_type, timestamp: e.timestamp, metadata: e.metadata ?? {}, + // Correlation envelope (#245): passed through per-event, omitted when absent. + ...(e.user_id && { user_id: e.user_id }), + ...(e.repo && { repo: e.repo }), + ...(e.trace_id && { trace_id: e.trace_id }), })); // Verification is non-null only when at least one gate result was persisted. diff --git a/cdk/src/handlers/orchestrate-task.ts b/cdk/src/handlers/orchestrate-task.ts index 2e7be6de..bfba0754 100644 --- a/cdk/src/handlers/orchestrate-task.ts +++ b/cdk/src/handlers/orchestrate-task.ts @@ -26,6 +26,7 @@ import { logger } from './shared/logger'; import { admissionControl, emitTaskEvent, + envelopeFor, failTask, finalizeTask, hydrateAndTransition, @@ -58,12 +59,18 @@ const durableHandler: DurableExecutionHandler = asyn return loadTask(taskId); }); + // Correlation envelope (#245): stamp {task_id, user_id, repo} on every + // lifecycle log line so admission→terminal transitions join to TaskEvents + // and the agent trace without hand-adding fields per call. `repo` is + // undefined for repo-less workflows (#248 Phase 3). + const { log, correlation } = envelopeFor(task); + // Step 1b: Load blueprint config (per-repo overrides) const blueprintConfig = await context.step('load-blueprint', async () => { try { return await loadBlueprintConfig(task); } catch (err) { - await failTask(taskId, task.status, `Blueprint config load failed: ${String(err)}`, task.user_id, false); + await failTask(taskId, task.status, `Blueprint config load failed: ${String(err)}`, task.user_id, false, task.repo); throw err; } }); @@ -77,23 +84,21 @@ const durableHandler: DurableExecutionHandler = asyn } const result = await admissionControl(task); if (!result) { - await failTask(taskId, current.status, 'User concurrency limit reached', task.user_id, false); - await emitTaskEvent(taskId, 'admission_rejected', { reason: 'concurrency_limit' }); + await failTask(taskId, current.status, 'User concurrency limit reached', task.user_id, false, task.repo); + await emitTaskEvent(taskId, 'admission_rejected', { reason: 'concurrency_limit' }, correlation); // Channel feedback is non-fatal: a throw here would re-run failTask + // emitTaskEvent on the durable-execution retry, producing duplicate events. try { await notifyLinearOnConcurrencyCap(task); } catch (err) { - logger.warn('Linear concurrency-cap feedback failed (non-fatal)', { - task_id: taskId, + log.warn('Linear concurrency-cap feedback failed (non-fatal)', { error: err instanceof Error ? err.message : String(err), }); } try { await notifyJiraOnConcurrencyCap(task); } catch (err) { - logger.warn('Jira concurrency-cap feedback failed (non-fatal)', { - task_id: taskId, + log.warn('Jira concurrency-cap feedback failed (non-fatal)', { error: err instanceof Error ? err.message : String(err), }); } @@ -122,16 +127,16 @@ const durableHandler: DurableExecutionHandler = asyn ); if (!result.passed) { const errorMessage = `Pre-flight check failed: ${result.failureReason}${result.failureDetail ? ' — ' + result.failureDetail : ''}`; - await failTask(taskId, current.status, errorMessage, task.user_id, true); + await failTask(taskId, current.status, errorMessage, task.user_id, true, task.repo); await emitTaskEvent(taskId, 'preflight_failed', { reason: result.failureReason, detail: result.failureDetail, checks: result.checks, - }); + }, correlation); } return result.passed; } catch (err) { - await failTask(taskId, task.status, `Pre-flight failed: ${String(err)}`, task.user_id, true); + await failTask(taskId, task.status, `Pre-flight failed: ${String(err)}`, task.user_id, true, task.repo); throw err; } }); @@ -146,7 +151,7 @@ const durableHandler: DurableExecutionHandler = asyn return await hydrateAndTransition(task, blueprintConfig); } catch (err) { // Hydration may fail due to external cancellation, guardrail blocking, or guardrail API failure — fail the task and release concurrency - await failTask(taskId, TaskStatus.HYDRATING, `Hydration failed: ${String(err)}`, task.user_id, true); + await failTask(taskId, TaskStatus.HYDRATING, `Hydration failed: ${String(err)}`, task.user_id, true, task.repo); throw err; } }); @@ -178,17 +183,16 @@ const durableHandler: DurableExecutionHandler = asyn await emitTaskEvent(taskId, 'session_started', { session_id: handle.sessionId, strategy_type: handle.strategyType, - }); + }, correlation); - logger.info('Session started', { - task_id: taskId, + log.info('Session started', { session_id: handle.sessionId, strategy_type: handle.strategyType, }); return handle; } catch (err) { - await failTask(taskId, TaskStatus.HYDRATING, `Session start failed: ${String(err)}`, task.user_id, true); + await failTask(taskId, TaskStatus.HYDRATING, `Session start failed: ${String(err)}`, task.user_id, true, task.repo); throw err; } }); @@ -223,11 +227,10 @@ const durableHandler: DurableExecutionHandler = asyn const ecsStatus = await computeStrategy.pollSession(sessionHandle); if (ecsStatus.status === 'failed') { const errorMsg = 'error' in ecsStatus ? ecsStatus.error : 'ECS task failed'; - logger.warn('ECS task failed before DDB terminal write', { - task_id: taskId, + log.warn('ECS task failed before DDB terminal write', { error: errorMsg, }); - await failTask(taskId, ddbState.lastStatus, `ECS container failed: ${errorMsg}`, task.user_id, false); + await failTask(taskId, ddbState.lastStatus, `ECS container failed: ${errorMsg}`, task.user_id, false, task.repo); return { attempts: ddbState.attempts, lastStatus: TaskStatus.FAILED }; } if (ecsStatus.status === 'completed') { @@ -235,31 +238,27 @@ const durableHandler: DurableExecutionHandler = asyn if (consecutiveEcsCompletedPolls >= MAX_CONSECUTIVE_ECS_COMPLETED_POLLS) { // ECS task exited successfully but DDB never reached terminal — the agent // likely crashed after container exit code 0 but before writing status. - logger.error('ECS task completed but DDB never caught up — failing task', { - task_id: taskId, + log.error('ECS task completed but DDB never caught up — failing task', { consecutive_completed_polls: consecutiveEcsCompletedPolls, }); - await failTask(taskId, ddbState.lastStatus, `ECS task exited successfully but agent never wrote terminal status after ${consecutiveEcsCompletedPolls} polls`, task.user_id, false); + await failTask(taskId, ddbState.lastStatus, `ECS task exited successfully but agent never wrote terminal status after ${consecutiveEcsCompletedPolls} polls`, task.user_id, false, task.repo); return { attempts: ddbState.attempts, lastStatus: TaskStatus.FAILED }; } - logger.warn('ECS task completed but DDB not terminal — waiting for DDB catchup', { - task_id: taskId, + log.warn('ECS task completed but DDB not terminal — waiting for DDB catchup', { consecutive_completed_polls: consecutiveEcsCompletedPolls, }); } } catch (err) { consecutiveEcsPollFailures = (state.consecutiveEcsPollFailures ?? 0) + 1; if (consecutiveEcsPollFailures >= MAX_CONSECUTIVE_ECS_POLL_FAILURES) { - logger.error('ECS pollSession failed repeatedly — failing task', { - task_id: taskId, + log.error('ECS pollSession failed repeatedly — failing task', { consecutive_failures: consecutiveEcsPollFailures, error: err instanceof Error ? err.message : String(err), }); - await failTask(taskId, ddbState.lastStatus, `ECS poll failed ${consecutiveEcsPollFailures} consecutive times: ${err instanceof Error ? err.message : String(err)}`, task.user_id, false); + await failTask(taskId, ddbState.lastStatus, `ECS poll failed ${consecutiveEcsPollFailures} consecutive times: ${err instanceof Error ? err.message : String(err)}`, task.user_id, false, task.repo); return { attempts: ddbState.attempts, lastStatus: TaskStatus.FAILED }; } - logger.warn('ECS pollSession check failed (non-fatal)', { - task_id: taskId, + log.warn('ECS pollSession check failed (non-fatal)', { consecutive_failures: consecutiveEcsPollFailures, error: err instanceof Error ? err.message : String(err), }); diff --git a/cdk/src/handlers/shared/logger.ts b/cdk/src/handlers/shared/logger.ts index 6fc379e0..a8d77fbc 100644 --- a/cdk/src/handlers/shared/logger.ts +++ b/cdk/src/handlers/shared/logger.ts @@ -17,36 +17,46 @@ * SOFTWARE. */ +/** Structured logger surface: the three level methods plus `child`. */ +export interface Logger { + info(message: string, data?: Record): void; + warn(message: string, data?: Record): void; + error(message: string, data?: Record): void; + /** + * Return a logger that merges `context` (e.g. the correlation envelope + * `{ task_id, user_id, repo }`) into every line, so per-call sites don't + * repeat it. Per-call `data` wins on key collision. + */ + child(context: Record): Logger; +} + /** * Minimal structured logger that writes JSON to stdout/stderr. * Uses process.stdout.write / process.stderr.write to avoid the * `no-console` eslint rule. CloudWatch captures both streams. + * + * @param context - persistent fields merged into every line (see `child`). */ -export const logger = { - /** - * Log an informational message. - * @param message - the log message. - * @param data - optional structured data to include. - */ - info(message: string, data?: Record): void { - process.stdout.write(JSON.stringify({ level: 'INFO', message, ...data }) + '\n'); - }, +// Closure over a context obj, not Powertools — no child-of-child depth or +// sampling needed, just a persistent correlation envelope. +function makeLogger(context: Record = {}): Logger { + const write = (stream: NodeJS.WriteStream, level: string, message: string, data?: Record): void => { + stream.write(JSON.stringify({ level, message, ...context, ...data }) + '\n'); + }; + return { + info(message, data) { + write(process.stdout, 'INFO', message, data); + }, + warn(message, data) { + write(process.stdout, 'WARN', message, data); + }, + error(message, data) { + write(process.stderr, 'ERROR', message, data); + }, + child(childContext) { + return makeLogger({ ...context, ...childContext }); + }, + }; +} - /** - * Log a warning message. - * @param message - the log message. - * @param data - optional structured data to include. - */ - warn(message: string, data?: Record): void { - process.stdout.write(JSON.stringify({ level: 'WARN', message, ...data }) + '\n'); - }, - - /** - * Log an error message. - * @param message - the log message. - * @param data - optional structured data to include. - */ - error(message: string, data?: Record): void { - process.stderr.write(JSON.stringify({ level: 'ERROR', message, ...data }) + '\n'); - }, -}; +export const logger: Logger = makeLogger(); diff --git a/cdk/src/handlers/shared/orchestrator.ts b/cdk/src/handlers/shared/orchestrator.ts index 27a8a0a6..7308c453 100644 --- a/cdk/src/handlers/shared/orchestrator.ts +++ b/cdk/src/handlers/shared/orchestrator.ts @@ -22,7 +22,7 @@ import { S3Client } from '@aws-sdk/client-s3'; import { DynamoDBDocumentClient, GetCommand, PutCommand, UpdateCommand } from '@aws-sdk/lib-dynamodb'; import { ulid } from 'ulid'; import { AttachmentBudgetExceededError, AttachmentConfigurationError, AttachmentResolutionError, hydrateContext, resolveGitHubToken } from './context-hydration'; -import { logger } from './logger'; +import { logger, type Logger } from './logger'; import { writeMinimalEpisode } from './memory'; import { coerceNumericOrNull } from './numeric'; import { computePromptVersion } from './prompt-version'; @@ -168,16 +168,45 @@ export async function transitionTask( })); } +/** Correlation envelope stamped on orchestrator-plane task events (#245). */ +export interface EventCorrelation { + readonly user_id?: string; + /** `owner/repo`, or absent for repo-less workflows (#248 Phase 3). Matches + * the source `TaskRecord.repo` (`string | undefined`). */ + readonly repo?: string; +} + +/** + * Build the correlation envelope (#245) for a task from its identity fields: + * a `log` child stamping `{task_id, user_id, repo}` on every line, and the + * matching `correlation` for `emitTaskEvent`. Single source so the log context + * and the event envelope can't drift. `repo` is omitted (not null/empty) for + * repo-less workflows. + */ +export function envelopeFor( + identity: { task_id: string; user_id: string; repo?: string }, +): { log: Logger; correlation: EventCorrelation } { + const { task_id, user_id, repo } = identity; + return { + log: logger.child({ task_id, user_id, ...(repo && { repo }) }), + correlation: { user_id, repo }, + }; +} + /** * Emit a task event to the audit log. * @param taskId - the task ID. * @param eventType - the event type string. * @param metadata - optional event metadata. + * @param correlation - optional `{ user_id, repo }` stamped as top-level fields + * so the event stream joins to orchestrator logs by the correlation envelope + * (#245). `repo` is omitted when null/absent (repo-less workflows). */ export async function emitTaskEvent( taskId: string, eventType: string, metadata?: Record, + correlation?: EventCorrelation, ): Promise { await ddb.send(new PutCommand({ TableName: EVENTS_TABLE_NAME, @@ -187,6 +216,8 @@ export async function emitTaskEvent( event_type: eventType, timestamp: new Date().toISOString(), ttl: computeTtlEpoch(TASK_RETENTION_DAYS), + ...(correlation?.user_id && { user_id: correlation.user_id }), + ...(correlation?.repo && { repo: correlation.repo }), ...(metadata && { metadata }), }, })); @@ -203,24 +234,23 @@ const MAX_POLL_INTERVAL_MS = 300_000; * @returns the merged blueprint config. */ export async function loadBlueprintConfig(task: TaskRecord): Promise { + // Correlation envelope (#245) on this shared function's logs too, matching the + // orchestrate handler's child logger. `repo` omitted for repo-less workflows. + const { log } = envelopeFor(task); + // Repo-less workflows (#248 Phase 3) have no per-repo Blueprint — use platform // defaults directly rather than a RepoTable lookup on a missing repo. const repoConfig = task.repo ? await loadRepoConfig(task.repo) : null; if (repoConfig) { - logger.info('Loaded per-repo blueprint config', { - task_id: task.task_id, - repo: task.repo, + log.info('Loaded per-repo blueprint config', { has_runtime_override: !!repoConfig.runtime_arn, has_model_override: !!repoConfig.model_id, has_prompt_override: !!repoConfig.system_prompt_overrides, has_token_override: !!repoConfig.github_token_secret_arn, }); } else { - logger.info('No per-repo config found, using platform defaults', { - task_id: task.task_id, - repo: task.repo, - }); + log.info('No per-repo config found, using platform defaults'); } // Clamp poll_interval_ms to safe range @@ -228,8 +258,7 @@ export async function loadBlueprintConfig(task: TaskRecord): Promise> { + const { log, correlation } = envelopeFor(task); await transitionTask(task.task_id, TaskStatus.SUBMITTED, TaskStatus.HYDRATING); - await emitTaskEvent(task.task_id, 'hydration_started'); + await emitTaskEvent(task.task_id, 'hydration_started', undefined, correlation); const hydratedContext = await hydrateContext(task, { githubTokenSecretArn: blueprintConfig?.github_token_secret_arn, @@ -350,10 +380,9 @@ export async function hydrateAndTransition(task: TaskRecord, blueprintConfig?: B sources: hydratedContext.sources, token_estimate: hydratedContext.token_estimate, ...(hydratedContext.content_trust && { content_trust: hydratedContext.content_trust }), - }); + }, correlation); } catch (eventErr) { - logger.error('Failed to emit guardrail_blocked event', { - task_id: task.task_id, + log.error('Failed to emit guardrail_blocked event', { error: eventErr instanceof Error ? eventErr.message : String(eventErr), }); } @@ -371,8 +400,7 @@ export async function hydrateAndTransition(task: TaskRecord, blueprintConfig?: B ExpressionAttributeValues: { ':bn': hydratedContext.resolved_branch_name, ':now': new Date().toISOString() }, })); } catch (err) { - logger.error('Failed to update branch_name from PR head_ref — task record will show stale placeholder', { - task_id: task.task_id, + log.error('Failed to update branch_name from PR head_ref — task record will show stale placeholder', { resolved_branch: hydratedContext.resolved_branch_name, error: err instanceof Error ? err.message : String(err), }); @@ -394,8 +422,8 @@ export async function hydrateAndTransition(task: TaskRecord, blueprintConfig?: B ExpressionAttributeValues: { ':pv': promptVersion, ':now': new Date().toISOString() }, })); } catch (err) { - logger.warn('Failed to store prompt_version on task record', { - task_id: task.task_id, error: err instanceof Error ? err.message : String(err), + log.warn('Failed to store prompt_version on task record', { + error: err instanceof Error ? err.message : String(err), }); } @@ -413,8 +441,7 @@ export async function hydrateAndTransition(task: TaskRecord, blueprintConfig?: B task.approval_gate_cap !== undefined && !isValidApprovalGateCap(task.approval_gate_cap) ) { - logger.warn('TaskRecord.approval_gate_cap is not a valid integer in bounds; omitting from agent payload', { - task_id: task.task_id, + log.warn('TaskRecord.approval_gate_cap is not a valid integer in bounds; omitting from agent payload', { approval_gate_cap: task.approval_gate_cap, min: APPROVAL_GATE_CAP_MIN, max: APPROVAL_GATE_CAP_MAX, @@ -469,8 +496,7 @@ export async function hydrateAndTransition(task: TaskRecord, blueprintConfig?: B ); } - logger.warn('Failed to resolve GitHub token for URL attachment fetch (no GitHub URLs in batch — proceeding)', { - task_id: task.task_id, + log.warn('Failed to resolve GitHub token for URL attachment fetch (no GitHub URLs in batch — proceeding)', { error: err instanceof Error ? err.message : String(err), }); } @@ -564,8 +590,7 @@ export async function hydrateAndTransition(task: TaskRecord, blueprintConfig?: B }; if (hydratedContext.fallback_error) { - logger.warn('Context hydration fell back to minimal payload', { - task_id: task.task_id, + log.warn('Context hydration fell back to minimal payload', { fallback_error: hydratedContext.fallback_error, }); } @@ -578,7 +603,7 @@ export async function hydrateAndTransition(task: TaskRecord, blueprintConfig?: B has_memory_context: !!hydratedContext.memory_context, ...(hydratedContext.content_trust && { content_trust: hydratedContext.content_trust }), ...(hydratedContext.fallback_error && { fallback_error: hydratedContext.fallback_error }), - }); + }, correlation); return payload; } @@ -663,6 +688,9 @@ export async function finalizeTask( ): Promise { const task = await loadTask(taskId); const currentStatus = task.status; + // Correlation envelope (#245) on this function's own log lines too, not just + // the events it emits — admission→terminal logs must join by {user_id, repo}. + const { log, correlation } = envelopeFor(task); // Lost session: RUNNING but agent heartbeats stopped (crash/OOM) — fail fast if ( @@ -680,8 +708,7 @@ export async function finalizeTask( } catch (err) { // Task may have transitioned concurrently (e.g. agent wrote terminal status). // Re-read to avoid double-decrement or contradictory events. - logger.warn('Finalization transition to FAILED (heartbeat) failed, task may have transitioned concurrently', { - task_id: taskId, + log.warn('Finalization transition to FAILED (heartbeat) failed, task may have transitioned concurrently', { error: err instanceof Error ? err.message : String(err), }); } @@ -689,21 +716,21 @@ export async function finalizeTask( await emitTaskEvent(taskId, 'task_failed', { reason: 'agent_heartbeat_stale', poll_attempts: pollState.attempts, - }); + }, correlation); await decrementConcurrency(userId); } else { // Transition failed — re-read task to determine actual state. // If already terminal the block below will handle TTL + concurrency. const reread = await loadTask(taskId); if (TERMINAL_STATUSES.includes(reread.status)) { - logger.info('Heartbeat path: task already terminal after failed transition', { task_id: taskId, status: reread.status }); + log.info('Heartbeat path: task already terminal after failed transition', { status: reread.status }); await emitTaskEvent(taskId, `task_${reread.status.toLowerCase()}`, { final_status: reread.status, poll_attempts: pollState.attempts, - }); + }, correlation); await decrementConcurrency(userId); } else { - logger.warn('Heartbeat path: task in unexpected state after failed transition, releasing concurrency', { task_id: taskId, status: reread.status }); + log.warn('Heartbeat path: task in unexpected state after failed transition, releasing concurrency', { status: reread.status }); await decrementConcurrency(userId); } } @@ -712,7 +739,7 @@ export async function finalizeTask( // If the agent already wrote a terminal status, just finalize if (TERMINAL_STATUSES.includes(currentStatus)) { - logger.info('Task already in terminal state', { task_id: taskId, status: currentStatus }); + log.info('Task already in terminal state', { status: currentStatus }); try { await ddb.send(new UpdateCommand({ @@ -723,12 +750,12 @@ export async function finalizeTask( ExpressionAttributeValues: { ':ttl': computeTtlEpoch(TASK_RETENTION_DAYS) }, })); } catch (err) { - logger.warn('Failed to stamp TTL on terminal task', { task_id: taskId, error: err instanceof Error ? err.message : String(err) }); + log.warn('Failed to stamp TTL on terminal task', { error: err instanceof Error ? err.message : String(err) }); } // Memory fallback: if the agent did not write memory, write a minimal episode if (MEMORY_ID && !task.memory_written) { - logger.info('Agent did not write memory — writing fallback episode', { task_id: taskId }); + log.info('Agent did not write memory — writing fallback episode'); try { // Coerce at the shared helper rather than ``Number(...)`` so a // corrupt string ``cost_usd`` from the DDB Document client @@ -758,11 +785,10 @@ export async function finalizeTask( costUsd ?? undefined, ); if (!written) { - logger.warn('Fallback episode write returned false', { task_id: taskId }); + log.warn('Fallback episode write returned false'); } } catch (err) { - logger.warn('Fallback episode write threw unexpectedly (fail-open)', { - task_id: taskId, + log.warn('Fallback episode write threw unexpectedly (fail-open)', { error: err instanceof Error ? err.message : String(err), }); } @@ -771,7 +797,7 @@ export async function finalizeTask( await emitTaskEvent(taskId, `task_${currentStatus.toLowerCase()}`, { final_status: currentStatus, poll_attempts: pollState.attempts, - }); + }, correlation); await decrementConcurrency(userId); return; } @@ -796,14 +822,14 @@ export async function finalizeTask( }); } catch (err) { // Task may have transitioned concurrently — re-read and accept - logger.warn('Finalization transition failed, task may have transitioned concurrently', { task_id: taskId, error: err instanceof Error ? err.message : String(err) }); + log.warn('Finalization transition failed, task may have transitioned concurrently', { error: err instanceof Error ? err.message : String(err) }); } await emitTaskEvent(taskId, 'task_timed_out', { reason: currentStatus === TaskStatus.AWAITING_APPROVAL ? 'approval_poll_timeout' : 'poll_timeout', poll_attempts: pollState.attempts, - }); + }, correlation); await decrementConcurrency(userId); return; } @@ -817,18 +843,18 @@ export async function finalizeTask( error_message: 'Session never started — poll timeout exceeded while still HYDRATING', }); } catch (err) { - logger.warn('Finalization transition from HYDRATING failed, task may have transitioned concurrently', { task_id: taskId, error: err instanceof Error ? err.message : String(err) }); + log.warn('Finalization transition from HYDRATING failed, task may have transitioned concurrently', { error: err instanceof Error ? err.message : String(err) }); } await emitTaskEvent(taskId, 'task_failed', { reason: 'session_never_started', poll_attempts: pollState.attempts, - }); + }, correlation); await decrementConcurrency(userId); return; } // Unexpected state — log and release concurrency - logger.error('Unexpected task state during finalization', { task_id: taskId, status: currentStatus }); + log.error('Unexpected task state during finalization', { status: currentStatus }); await decrementConcurrency(userId); } @@ -839,6 +865,8 @@ export async function finalizeTask( * @param errorMessage - the error reason. * @param userId - the user who owns the task. * @param releaseConcurrency - whether to decrement the concurrency counter. + * @param repo - optional target repo (`owner/repo`) for the correlation + * envelope (#245); omit for repo-less workflows. */ export async function failTask( taskId: string, @@ -846,6 +874,7 @@ export async function failTask( errorMessage: string, userId: string, releaseConcurrency: boolean, + repo?: string, ): Promise { let transitioned = false; try { @@ -855,13 +884,16 @@ export async function failTask( }); transitioned = true; } catch (err) { - logger.warn('Failed to transition task to FAILED', { task_id: taskId, error: err instanceof Error ? err.message : String(err) }); + const { log } = envelopeFor({ task_id: taskId, user_id: userId, repo }); + log.warn('Failed to transition task to FAILED', { + error: err instanceof Error ? err.message : String(err), + }); } // Only emit / release concurrency after a successful transition. Callers such as // orchestrate-task rethrow after failTask; Durable Execution retries the step and // would otherwise re-run emit + decrement while the task is already FAILED. if (transitioned) { - await emitTaskEvent(taskId, 'task_failed', { error_message: errorMessage }); + await emitTaskEvent(taskId, 'task_failed', { error_message: errorMessage }, { user_id: userId, repo }); if (releaseConcurrency) { await decrementConcurrency(userId); } diff --git a/cdk/src/handlers/shared/types.ts b/cdk/src/handlers/shared/types.ts index 35d9a8a3..468a7c20 100644 --- a/cdk/src/handlers/shared/types.ts +++ b/cdk/src/handlers/shared/types.ts @@ -344,6 +344,11 @@ export interface EventRecord { readonly timestamp: string; readonly metadata?: Record; readonly ttl?: number; + // Correlation envelope (#245): present on events written after task creation; + // absent on `task_created` and any pre-envelope safety-net writer. + readonly user_id?: string; + readonly repo?: string; + readonly trace_id?: string; } /** @@ -358,6 +363,10 @@ export interface ReplayEvent { readonly event_type: string; readonly timestamp: string; readonly metadata: Record; + // Correlation envelope (#245): present per-event when the source stamped it. + readonly user_id?: string; + readonly repo?: string; + readonly trace_id?: string; } /** diff --git a/cdk/test/handlers/get-task-events.test.ts b/cdk/test/handlers/get-task-events.test.ts index ad8c70fa..f74a98f0 100644 --- a/cdk/test/handlers/get-task-events.test.ts +++ b/cdk/test/handlers/get-task-events.test.ts @@ -64,6 +64,10 @@ const EVENT_ITEMS = [ event_type: 'session_started', timestamp: '2025-03-15T10:31:00Z', metadata: { session_id: 'sess-1' }, + // Correlation envelope (#245) stamped on post-creation events. + user_id: 'user-123', + repo: 'org/repo', + trace_id: 'a'.repeat(32), }, ]; @@ -136,6 +140,21 @@ describe('get-task-events handler', () => { expect(body.pagination.has_more).toBe(false); }); + test('passes through the correlation envelope (#245), omitting absent fields', async () => { + const result = await handler(makeEvent()); + const body = JSON.parse(result.body); + // task_created predates the envelope → no correlation fields. + expect(body.data[0]).not.toHaveProperty('user_id'); + expect(body.data[0]).not.toHaveProperty('repo'); + expect(body.data[0]).not.toHaveProperty('trace_id'); + // session_started carries the full envelope. + expect(body.data[1]).toMatchObject({ + user_id: 'user-123', + repo: 'org/repo', + trace_id: 'a'.repeat(32), + }); + }); + test('returns 401 when user is not authenticated', async () => { const event = makeEvent(); event.requestContext.authorizer = null; diff --git a/cdk/test/handlers/get-task-replay.test.ts b/cdk/test/handlers/get-task-replay.test.ts index bfc11c3f..a1d5e865 100644 --- a/cdk/test/handlers/get-task-replay.test.ts +++ b/cdk/test/handlers/get-task-replay.test.ts @@ -316,6 +316,29 @@ describe('assembleBundle', () => { { event_id: '01B', event_type: 'agent_turn', timestamp: 'ts2', metadata: { turn: 1 } }, ]); }); + + test('passes through the correlation envelope (#245) per-event, omitting absent fields', () => { + const base = { task_id: 't', user_id: 'u' } as unknown as TaskRecord; + const evs = [ + // task_created predates the envelope → no correlation fields. + { task_id: 't', event_id: '01A', event_type: 'task_created', timestamp: 'ts' }, + // agent event carries the full envelope. + { + task_id: 't', + event_id: '01B', + event_type: 'agent_turn', + timestamp: 'ts2', + metadata: { turn: 1 }, + user_id: 'u', + repo: 'org/repo', + trace_id: 'a'.repeat(32), + }, + ] as unknown as EventRecord[]; + const out = assembleBundle(base, evs).events; + expect(out[0]).not.toHaveProperty('user_id'); + expect(out[0]).not.toHaveProperty('trace_id'); + expect(out[1]).toMatchObject({ user_id: 'u', repo: 'org/repo', trace_id: 'a'.repeat(32) }); + }); }); describe('replay-bundle.example.json fixture (#515 AC: example fixture in cdk/test/)', () => { diff --git a/cdk/test/handlers/orchestrate-task.test.ts b/cdk/test/handlers/orchestrate-task.test.ts index b6ec4e44..4314896c 100644 --- a/cdk/test/handlers/orchestrate-task.test.ts +++ b/cdk/test/handlers/orchestrate-task.test.ts @@ -70,6 +70,7 @@ process.env.MEMORY_ID = 'mem-test-default'; import { admissionControl, emitTaskEvent, + envelopeFor, failTask, finalizeTask, hydrateAndTransition, @@ -286,6 +287,91 @@ describe('hydrateAndTransition', () => { expect(metadata.truncated).toBe(false); expect(metadata.content_trust).toEqual({ task_description: 'trusted' }); }); + + test('lifecycle events carry the {user_id, repo} correlation envelope (#245)', async () => { + mockDdbSend.mockResolvedValue({}); + mockHydrateContext.mockResolvedValueOnce(mockHydratedContext); + await hydrateAndTransition(baseTask as any); + const events = mockDdbSend.mock.calls + .filter((c: any) => c[0]._type === 'Put') + .map((c: any) => c[0].input.Item); + for (const eventType of ['hydration_started', 'hydration_complete']) { + const evt = events.find((e: any) => e.event_type === eventType); + expect(evt).toMatchObject({ user_id: 'user-123', repo: 'org/repo' }); + } + }); + + test('repo-less task omits repo but still stamps user_id (#245, #248)', async () => { + mockDdbSend.mockResolvedValue({}); + mockHydrateContext.mockResolvedValueOnce(mockHydratedContext); + const { repo: _repo, ...repoLess } = baseTask; + await hydrateAndTransition(repoLess as any); + const started = mockDdbSend.mock.calls + .filter((c: any) => c[0]._type === 'Put') + .map((c: any) => c[0].input.Item) + .find((e: any) => e.event_type === 'hydration_started'); + expect(started.user_id).toBe('user-123'); + expect(started).not.toHaveProperty('repo'); + }); +}); + +describe('emitTaskEvent — correlation envelope (#245)', () => { + test('stamps user_id and repo as top-level fields when supplied', async () => { + mockDdbSend.mockResolvedValue({}); + await emitTaskEvent('TASK001', 'session_started', { session_id: 's-1' }, { user_id: 'user-123', repo: 'org/repo' }); + const item = mockDdbSend.mock.calls.find((c: any) => c[0]._type === 'Put')[0].input.Item; + expect(item).toMatchObject({ + task_id: 'TASK001', + event_type: 'session_started', + user_id: 'user-123', + repo: 'org/repo', + }); + expect(item.metadata).toEqual({ session_id: 's-1' }); + }); + + test('omits repo (not null/empty) when repo-less, keeps user_id', async () => { + mockDdbSend.mockResolvedValue({}); + await emitTaskEvent('TASK001', 'task_failed', undefined, { user_id: 'user-123', repo: undefined }); + const item = mockDdbSend.mock.calls.find((c: any) => c[0]._type === 'Put')[0].input.Item; + expect(item.user_id).toBe('user-123'); + expect(item).not.toHaveProperty('repo'); + }); + + test('omits both when no correlation is supplied (back-compat)', async () => { + mockDdbSend.mockResolvedValue({}); + await emitTaskEvent('TASK001', 'task_created'); + const item = mockDdbSend.mock.calls.find((c: any) => c[0]._type === 'Put')[0].input.Item; + expect(item).not.toHaveProperty('user_id'); + expect(item).not.toHaveProperty('repo'); + }); +}); + +describe('envelopeFor — single source for log context + event correlation (#245)', () => { + function captureLine(fn: () => void): Record { + const lines: string[] = []; + const spy = jest.spyOn(process.stdout, 'write').mockImplementation((chunk: unknown) => { + lines.push(String(chunk)); + return true; + }); + try { fn(); } finally { spy.mockRestore(); } + return JSON.parse(lines[0]) as Record; + } + + test('log child and correlation both carry user_id + repo', () => { + const { log, correlation } = envelopeFor({ task_id: 't-1', user_id: 'u-1', repo: 'o/r' }); + // Regression guard for the loadBlueprintConfig gap: the log child must carry + // user_id, not just task_id/repo — every admission→terminal line needs it. + expect(captureLine(() => log.info('x'))).toMatchObject({ task_id: 't-1', user_id: 'u-1', repo: 'o/r' }); + expect(correlation).toEqual({ user_id: 'u-1', repo: 'o/r' }); + }); + + test('repo-less: repo omitted from the log line and left undefined on correlation', () => { + const { log, correlation } = envelopeFor({ task_id: 't-1', user_id: 'u-1' }); + const line = captureLine(() => log.info('x')); + expect(line).toMatchObject({ task_id: 't-1', user_id: 'u-1' }); + expect(line).not.toHaveProperty('repo'); + expect(correlation.repo).toBeUndefined(); + }); }); describe('hydrateAndTransition — Cedar HITL payload threading', () => { @@ -808,6 +894,8 @@ describe('finalizeTask', () => { const eventCall = mockDdbSend.mock.calls[2][0]; expect(eventCall.input.Item.event_type).toBe('task_failed'); expect(eventCall.input.Item.metadata.reason).toBe('agent_heartbeat_stale'); + // Terminal event carries the correlation envelope (#245). + expect(eventCall.input.Item).toMatchObject({ user_id: 'user-123', repo: 'org/repo' }); }); test('transitions RUNNING to TIMED_OUT on poll timeout', async () => { diff --git a/cdk/test/handlers/shared/logger.test.ts b/cdk/test/handlers/shared/logger.test.ts new file mode 100644 index 00000000..d14c6eac --- /dev/null +++ b/cdk/test/handlers/shared/logger.test.ts @@ -0,0 +1,70 @@ +/** + * MIT No Attribution + * + * Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. + * + * Permission is hereby granted, free of charge, to any person obtaining a copy of + * the Software without restriction, including without limitation the rights to + * use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of + * the Software, and to permit persons to whom the Software is furnished to do so. + * + * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR + * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, + * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE + * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER + * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, + * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE + * SOFTWARE. + */ + +import { logger } from '../../../src/handlers/shared/logger'; + +/** Capture the first JSON line written to a stream during `fn`. */ +function captureLine(stream: 'stdout' | 'stderr', fn: () => void): Record { + const lines: string[] = []; + const spy = jest.spyOn(process[stream], 'write').mockImplementation((chunk: unknown) => { + lines.push(String(chunk)); + return true; + }); + try { + fn(); + } finally { + spy.mockRestore(); + } + return JSON.parse(lines[0]) as Record; +} + +describe('logger', () => { + test('info/warn write to stdout, error to stderr', () => { + expect(captureLine('stdout', () => logger.info('hi')).level).toBe('INFO'); + expect(captureLine('stdout', () => logger.warn('hi')).level).toBe('WARN'); + expect(captureLine('stderr', () => logger.error('hi')).level).toBe('ERROR'); + }); + + test('child merges persistent context into every line (correlation envelope)', () => { + const log = logger.child({ task_id: 't-1', user_id: 'u-1', repo: 'o/r' }); + const line = captureLine('stdout', () => log.info('Session started', { session_id: 's-1' })); + expect(line).toMatchObject({ + level: 'INFO', + message: 'Session started', + task_id: 't-1', + user_id: 'u-1', + repo: 'o/r', + session_id: 's-1', + }); + }); + + test('per-call data wins over child context on key collision', () => { + const log = logger.child({ task_id: 't-1', repo: 'o/r' }); + const line = captureLine('stdout', () => log.warn('override', { repo: 'other/repo' })); + expect(line.repo).toBe('other/repo'); + }); + + test('child is composable and does not mutate the base logger', () => { + const withTask = logger.child({ task_id: 't-1' }); + const withUser = withTask.child({ user_id: 'u-1' }); + expect(captureLine('stdout', () => withUser.info('x'))).toMatchObject({ task_id: 't-1', user_id: 'u-1' }); + // Base logger stays context-free. + expect(captureLine('stdout', () => logger.info('x')).task_id).toBeUndefined(); + }); +}); diff --git a/cli/src/types.ts b/cli/src/types.ts index bf397493..3ee7f91f 100644 --- a/cli/src/types.ts +++ b/cli/src/types.ts @@ -182,6 +182,10 @@ export interface ReplayEvent { readonly event_type: string; readonly timestamp: string; readonly metadata: Record; + // Correlation envelope (#245): present per-event when the source stamped it. + readonly user_id?: string; + readonly repo?: string; + readonly trace_id?: string; } /** @@ -236,6 +240,11 @@ export interface TaskEvent { readonly event_type: string; readonly timestamp: string; readonly metadata: Record; + // Correlation envelope (#245): present per-event when the source stamped it; + // absent on task_created and pre-envelope safety-net writers. + readonly user_id?: string; + readonly repo?: string; + readonly trace_id?: string; } /** diff --git a/docs/design/API_CONTRACT.md b/docs/design/API_CONTRACT.md index 2d06402d..9aa24135 100644 --- a/docs/design/API_CONTRACT.md +++ b/docs/design/API_CONTRACT.md @@ -329,6 +329,8 @@ Returns the audit trail for a task: state transitions, hydration events, session **Event types:** `task_created`, `uploads_confirmed`, `attachment_blocked`, `admission_rejected`, `preflight_failed`, `hydration_started`, `hydration_complete`, `session_started`, `pr_created`, `task_completed`, `task_failed`, `task_cancelled`, `task_timed_out`. Custom blueprint steps emit `{step_name}_started`, `{step_name}_completed`, `{step_name}_failed`. (There is no `admission_passed` or `session_ended` event — admission success is implicit in the subsequent `hydration_started`, and session end is captured by the terminal event.) +**Correlation envelope (#245):** each event optionally carries top-level `user_id`, `repo`, and `trace_id` (the agent's OTEL trace id) so the event stream joins to structured logs and X-Ray. Each field is present only when the event's source stamped it — `task_created` predates the envelope and carries none; events from `hydration_started` onward carry `user_id`/`repo`, and agent-emitted events additionally carry `trace_id`. Fields are omitted (not `null`) when absent. See [OBSERVABILITY.md](./OBSERVABILITY.md#correlation-envelope). + **Errors:** `401 UNAUTHORIZED`, `403 FORBIDDEN`, `404 TASK_NOT_FOUND`. ### Approve / deny a Cedar HITL gate @@ -414,7 +416,7 @@ Fields sourced from stores that may not have run for a given task are returned a - `otel_trace_id` is `null` on tasks created before the field existed or that ran with tracing unavailable; `session_id` is the available correlation id in that case. - `verification` is `null` when no gate result was persisted (e.g. a repo-less workflow has no build/lint step). - The embedded `events` list is capped both by count (`MAX_REPLAY_EVENTS`) and by total size (`MAX_REPLAY_EVENT_BYTES`, to stay under Lambda's 6 MB response limit — relevant for `--trace` tasks whose events carry large previews); whichever cap trips first truncates the tail. When that happens, `events_truncation` is non-null (`{ reason: "max_events" | "max_bytes", returned_events }`) so a consumer can detect a partial list; it is `null` when the full list fit. Use `GET /v1/tasks/{task_id}/events` for the full paginated feed. -- Each embedded event has the same shape as the `/events` feed — `event_id`, `event_type`, `timestamp`, and `metadata` (always present, `{}` when the event stored none); the internal `task_id`/`ttl` are stripped. +- Each embedded event has the same shape as the `/events` feed — `event_id`, `event_type`, `timestamp`, `metadata` (always present, `{}` when the event stored none), and the optional correlation envelope (`user_id`, `repo`, `trace_id`) when the source stamped it; the internal `task_id`/`ttl` are stripped. **Response: `200 OK`** diff --git a/docs/design/OBSERVABILITY.md b/docs/design/OBSERVABILITY.md index a8e9bbce..b6db7a8d 100644 --- a/docs/design/OBSERVABILITY.md +++ b/docs/design/OBSERVABILITY.md @@ -53,6 +53,37 @@ Root span attributes (`task.id`, `repo.url`, `agent.model`, `agent.cost_usd`, `b **Session correlation**: the AgentCore session ID propagates via OTEL baggage, linking custom spans to AgentCore's built-in session metrics in the CloudWatch GenAI Observability dashboard. +## Correlation envelope + +Every action in a task's lifecycle is attributable to a stable set of fields — the **correlation envelope** — so operators can answer "who triggered this, on which repo, in which task?" by joining CloudWatch logs, the `TaskEvents` table, and X-Ray traces on shared keys rather than manually stitching them. + +| Field | Meaning | Authoritative source | Optional? | +|-------|---------|----------------------|-----------| +| `task_id` | ULID identifying the task | `createTask` (API) → `TaskRecord.task_id` | No — present on every plane | +| `user_id` | Platform identity (Cognito `sub`) that triggered the task | `TaskRecord.user_id`, set at task creation | No | +| `repo` | Target repository (`owner/repo`) | `TaskRecord.repo` | **Yes** — **absent** (key omitted, not `null`/`""`) for repo-less workflows (`requires_repo: false`, #248 Phase 3). Consumers must treat it as optional; the memory/attribution fallback namespace is `user:{user_id}` | +| `trace_id` | OTEL trace id (32-char lowercase hex) of the agent run | agent's active root span, via `current_otel_trace_id()`; persisted to `TaskRecord.otel_trace_id` | Yes — `null` when tracing is disabled or the span context is invalid | +| `session_id` | AgentCore session id | compute strategy at `start-session`; propagated via OTEL baggage | Yes — absent until a session starts; used as the `trace_id` proxy when the trace id is unavailable | + +**Field naming is snake_case and shared** with Bedrock cost attribution (#215), the compliance export schema (#237), and delegation-chain propagation (#249), so the same names join across all four features. + +### Join model + +The orchestrator runs *before* the agent's trace exists, so there is no single W3C trace root spanning both planes. Instead, the `TaskRecord` is the orchestrator↔agent bridge: + +``` +orchestrator log ──task_id──▶ TaskRecord ──otel_trace_id──▶ agent trace + TaskEvents + S3 trace artifact +``` + +- **Orchestrator plane** (Lambda): structured logs and `TaskEvents` carry `{task_id, user_id, repo}`. The orchestrator never sees `trace_id` (the agent trace hasn't started), so it joins to the agent plane through the `TaskRecord`. +- **Agent plane** (MicroVM): OTel spans, baggage, and agent-emitted `TaskEvents` carry `{task_id, user_id, repo, trace_id}`, so the event stream is joinable to the X-Ray trace directly. +- **TaskRecord**: holds `otel_trace_id` and `session_id` (persisted at terminal write, #515), the durable link between the two planes and the basis of the [task replay bundle](#task-replay-bundle). + +**Coverage notes:** + +- The `task_created` event is written at task creation, *before* the correlation envelope exists on the orchestration path; it joins by `task_id` alone. Every event from `hydration_started` onward carries the envelope. Rare safety-net writers (e.g. the stranded-task reconciler) likewise join by `task_id`. +- The envelope is stamped on the stored `TaskEvents` items and is queryable directly in DynamoDB / CloudWatch. The `GET /tasks/{id}/events` and `/replay` API responses surface `user_id`/`repo`/`trace_id` per event when present; CLI consumers (`bgagent watch`/`events`/`replay`) can also join at the task level via `TaskRecord.otel_trace_id` (exposed on the replay bundle). + ## What to observe The platform tracks four categories of signals, each serving different consumers (operators, users, evaluation pipeline). diff --git a/docs/src/content/docs/architecture/Api-contract.md b/docs/src/content/docs/architecture/Api-contract.md index 827acf9a..458773eb 100644 --- a/docs/src/content/docs/architecture/Api-contract.md +++ b/docs/src/content/docs/architecture/Api-contract.md @@ -333,6 +333,8 @@ Returns the audit trail for a task: state transitions, hydration events, session **Event types:** `task_created`, `uploads_confirmed`, `attachment_blocked`, `admission_rejected`, `preflight_failed`, `hydration_started`, `hydration_complete`, `session_started`, `pr_created`, `task_completed`, `task_failed`, `task_cancelled`, `task_timed_out`. Custom blueprint steps emit `{step_name}_started`, `{step_name}_completed`, `{step_name}_failed`. (There is no `admission_passed` or `session_ended` event — admission success is implicit in the subsequent `hydration_started`, and session end is captured by the terminal event.) +**Correlation envelope (#245):** each event optionally carries top-level `user_id`, `repo`, and `trace_id` (the agent's OTEL trace id) so the event stream joins to structured logs and X-Ray. Each field is present only when the event's source stamped it — `task_created` predates the envelope and carries none; events from `hydration_started` onward carry `user_id`/`repo`, and agent-emitted events additionally carry `trace_id`. Fields are omitted (not `null`) when absent. See [OBSERVABILITY.md](/sample-autonomous-cloud-coding-agents/architecture/observability#correlation-envelope). + **Errors:** `401 UNAUTHORIZED`, `403 FORBIDDEN`, `404 TASK_NOT_FOUND`. ### Approve / deny a Cedar HITL gate @@ -418,7 +420,7 @@ Fields sourced from stores that may not have run for a given task are returned a - `otel_trace_id` is `null` on tasks created before the field existed or that ran with tracing unavailable; `session_id` is the available correlation id in that case. - `verification` is `null` when no gate result was persisted (e.g. a repo-less workflow has no build/lint step). - The embedded `events` list is capped both by count (`MAX_REPLAY_EVENTS`) and by total size (`MAX_REPLAY_EVENT_BYTES`, to stay under Lambda's 6 MB response limit — relevant for `--trace` tasks whose events carry large previews); whichever cap trips first truncates the tail. When that happens, `events_truncation` is non-null (`{ reason: "max_events" | "max_bytes", returned_events }`) so a consumer can detect a partial list; it is `null` when the full list fit. Use `GET /v1/tasks/{task_id}/events` for the full paginated feed. -- Each embedded event has the same shape as the `/events` feed — `event_id`, `event_type`, `timestamp`, and `metadata` (always present, `{}` when the event stored none); the internal `task_id`/`ttl` are stripped. +- Each embedded event has the same shape as the `/events` feed — `event_id`, `event_type`, `timestamp`, `metadata` (always present, `{}` when the event stored none), and the optional correlation envelope (`user_id`, `repo`, `trace_id`) when the source stamped it; the internal `task_id`/`ttl` are stripped. **Response: `200 OK`** diff --git a/docs/src/content/docs/architecture/Observability.md b/docs/src/content/docs/architecture/Observability.md index 714d1302..e032fd65 100644 --- a/docs/src/content/docs/architecture/Observability.md +++ b/docs/src/content/docs/architecture/Observability.md @@ -57,6 +57,37 @@ Root span attributes (`task.id`, `repo.url`, `agent.model`, `agent.cost_usd`, `b **Session correlation**: the AgentCore session ID propagates via OTEL baggage, linking custom spans to AgentCore's built-in session metrics in the CloudWatch GenAI Observability dashboard. +## Correlation envelope + +Every action in a task's lifecycle is attributable to a stable set of fields — the **correlation envelope** — so operators can answer "who triggered this, on which repo, in which task?" by joining CloudWatch logs, the `TaskEvents` table, and X-Ray traces on shared keys rather than manually stitching them. + +| Field | Meaning | Authoritative source | Optional? | +|-------|---------|----------------------|-----------| +| `task_id` | ULID identifying the task | `createTask` (API) → `TaskRecord.task_id` | No — present on every plane | +| `user_id` | Platform identity (Cognito `sub`) that triggered the task | `TaskRecord.user_id`, set at task creation | No | +| `repo` | Target repository (`owner/repo`) | `TaskRecord.repo` | **Yes** — **absent** (key omitted, not `null`/`""`) for repo-less workflows (`requires_repo: false`, #248 Phase 3). Consumers must treat it as optional; the memory/attribution fallback namespace is `user:{user_id}` | +| `trace_id` | OTEL trace id (32-char lowercase hex) of the agent run | agent's active root span, via `current_otel_trace_id()`; persisted to `TaskRecord.otel_trace_id` | Yes — `null` when tracing is disabled or the span context is invalid | +| `session_id` | AgentCore session id | compute strategy at `start-session`; propagated via OTEL baggage | Yes — absent until a session starts; used as the `trace_id` proxy when the trace id is unavailable | + +**Field naming is snake_case and shared** with Bedrock cost attribution (#215), the compliance export schema (#237), and delegation-chain propagation (#249), so the same names join across all four features. + +### Join model + +The orchestrator runs *before* the agent's trace exists, so there is no single W3C trace root spanning both planes. Instead, the `TaskRecord` is the orchestrator↔agent bridge: + +``` +orchestrator log ──task_id──▶ TaskRecord ──otel_trace_id──▶ agent trace + TaskEvents + S3 trace artifact +``` + +- **Orchestrator plane** (Lambda): structured logs and `TaskEvents` carry `{task_id, user_id, repo}`. The orchestrator never sees `trace_id` (the agent trace hasn't started), so it joins to the agent plane through the `TaskRecord`. +- **Agent plane** (MicroVM): OTel spans, baggage, and agent-emitted `TaskEvents` carry `{task_id, user_id, repo, trace_id}`, so the event stream is joinable to the X-Ray trace directly. +- **TaskRecord**: holds `otel_trace_id` and `session_id` (persisted at terminal write, #515), the durable link between the two planes and the basis of the [task replay bundle](#task-replay-bundle). + +**Coverage notes:** + +- The `task_created` event is written at task creation, *before* the correlation envelope exists on the orchestration path; it joins by `task_id` alone. Every event from `hydration_started` onward carries the envelope. Rare safety-net writers (e.g. the stranded-task reconciler) likewise join by `task_id`. +- The envelope is stamped on the stored `TaskEvents` items and is queryable directly in DynamoDB / CloudWatch. The `GET /tasks/{id}/events` and `/replay` API responses surface `user_id`/`repo`/`trace_id` per event when present; CLI consumers (`bgagent watch`/`events`/`replay`) can also join at the task level via `TaskRecord.otel_trace_id` (exposed on the replay bundle). + ## What to observe The platform tracks four categories of signals, each serving different consumers (operators, users, evaluation pipeline).