feat(carve S7): the orchestration compute plane — reconciler, release, rollup

cdk/src/constructs/orchestration-reconciler.ts

@@ -0,0 +1,176 @@
+/**
+ *  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 * as path from 'path';
+import { Duration } from 'aws-cdk-lib';
+import * as dynamodb from 'aws-cdk-lib/aws-dynamodb';
+import { Architecture, FilterCriteria, FilterRule, Runtime, StartingPosition } from 'aws-cdk-lib/aws-lambda';
+import { DynamoEventSource, SqsDlq } from 'aws-cdk-lib/aws-lambda-event-sources';
+import * as lambda from 'aws-cdk-lib/aws-lambda-nodejs';
+import * as sqs from 'aws-cdk-lib/aws-sqs';
+import { NagSuppressions } from 'cdk-nag';
+import { Construct } from 'constructs';
+import { TERMINAL_STATUSES } from './task-status';
+
+/**
+ * Properties for OrchestrationReconciler construct.
+ */
+export interface OrchestrationReconcilerProps {
+  /**
+   * TaskTable — MUST have a stream enabled (NEW_IMAGE). This construct is
+   * the table's stream consumer; the reconciler reacts to child tasks
+   * reaching terminal status.
+   */
+  readonly taskTable: dynamodb.ITable;
+
+  /** OrchestrationTable — the reconciler reads the DAG + writes child statuses. */
+  readonly orchestrationTable: dynamodb.ITable;
+
+  /** Orchestrator function ARN — releaseChild → createTaskCore invokes it. */
+  readonly orchestratorFunctionArn?: string;
+
+  /** Forwarded so released child tasks land in the right tables. */
+  readonly taskEventsTable: dynamodb.ITable;
+}
+
+/**
+ * TaskTable-stream consumer that drives Linear parent/sub-issue
+ * orchestration. On each child task reaching a
+ * terminal status it releases newly-unblocked children in dependency
+ * order (see `handlers/orchestration-reconciler.ts`).
+ *
+ * Stream-source rationale: TaskEventsTable's stream is at its 2-consumer
+ * limit (FanOutConsumer + ApprovalMetricsPublisher); TaskTable had no
+ * stream, so the reconciler is its first and only consumer — zero
+ * contention with the fan-out plane.
+ */
+
+/** DLQ message retention (days) — long enough for an operator to inspect a
+ *  poison stream record before it ages out. */
+const DLQ_RETENTION_DAYS = 14;
+
+export class OrchestrationReconciler extends Construct {
+  public readonly fn: lambda.NodejsFunction;
+  public readonly dlq: sqs.Queue;
+
+  constructor(scope: Construct, id: string, props: OrchestrationReconcilerProps) {
+    super(scope, id);
+
+    const handlersDir = path.join(__dirname, '..', 'handlers');
+
+    this.fn = new lambda.NodejsFunction(this, 'ReconcilerFn', {
+      entry: path.join(handlersDir, 'orchestration-reconciler.ts'),
+      handler: 'handler',
+      runtime: Runtime.NODEJS_24_X,
+      architecture: Architecture.ARM_64,
+      timeout: Duration.minutes(2),
+      // 512 MB (not 256): the reconciler bundles createTaskCore, which
+      // pulls in the Bedrock guardrail + S3 attachment-screening SDK
+      // stack. At 256 MB it OOMs during init on every stream event
+      // (Max Memory Used 255/256 MB) and never releases children. The
+      // LinearIntegration webhook processor runs the same code at 512 MB.
+      memorySize: 512,
+      environment: {
+        ORCHESTRATION_TABLE_NAME: props.orchestrationTable.tableName,
+        TASK_TABLE_NAME: props.taskTable.tableName,
+        TASK_EVENTS_TABLE_NAME: props.taskEventsTable.tableName,
+        ...(props.orchestratorFunctionArn && {
+          ORCHESTRATOR_FUNCTION_ARN: props.orchestratorFunctionArn,
+        }),
+      },
+      bundling: {
+        externalModules: ['@aws-sdk/*'],
+        // pdf-parse (v2, pdfjs-based) can't be esbuild-bundled — its pdfjs/native
+        // deps break at import. The reconciler screens the parent issue's PDF
+        // attachments, so ship pdf-parse
+        // unbundled to resolve natively at runtime. MUST match the webhook
+        // processors' attachment-screening bundling — a Lambda that CALLS
+        // attachment-screening but omits this carve-out fails every PDF at
+        // runtime while passing every unit test — a failure mode observed in
+        // practice on the attachment path.
+        nodeModules: ['pdf-parse'],
+      },
+    });
+
+    // DLQ for poison stream records (a record that repeatedly fails the
+    // reconcile). Fan-out uses the same pattern; without it a bad record
+    // would block the shard.
+    this.dlq = new sqs.Queue(this, 'ReconcilerDlq', {
+      retentionPeriod: Duration.days(DLQ_RETENTION_DAYS),
+      enforceSSL: true,
+    });
+
+    // Orchestration child creation/gating reads + writes the DAG table,
+    // reads/writes TaskTable (createTaskCore), and writes task events.
+    props.orchestrationTable.grantReadWriteData(this.fn);
+    props.taskTable.grantReadWriteData(this.fn);
+    props.taskEventsTable.grantReadWriteData(this.fn);
+
+    // Subscribe to the TaskTable stream. LATEST: we only care about
+    // tasks transitioning to terminal from here on. bisectBatchOnError +
+    // DLQ so one poison record can't wedge the shard.
+    //
+    // FilterCriteria: the handler ignores every non-terminal status
+    // (parseTerminalTaskRecord returns null unless status ∈ TERMINAL), so the
+    // stream itself filters to terminal statuses. This keeps RUNNING/HYDRATING/
+    // heartbeat/progress writes — the bulk of TaskTable churn platform-wide —
+    // from ever invoking this 512MB reconciler. Behavior-preserving: the records
+    // dropped here are exactly the ones the handler already discarded. One filter
+    // pattern per terminal status (FilterCriteria ORs the array).
+    const terminalFilters = TERMINAL_STATUSES.map((s) => FilterCriteria.filter({
+      dynamodb: { NewImage: { status: { S: FilterRule.isEqual(s) } } },
+    }));
+    this.fn.addEventSource(new DynamoEventSource(props.taskTable, {
+      startingPosition: StartingPosition.LATEST,
+      batchSize: 10,
+      retryAttempts: 3,
+      bisectBatchOnError: true,
+      onFailure: new SqsDlq(this.dlq),
+      filters: terminalFilters,
+      // Partial-batch reporting. Without this, the handler
+      // returning void meant ANY thrown record failed the WHOLE batch — a single
+      // poison/throttled record re-drove all its siblings (re-reconciling
+      // healthy children) until it aged out. With reportBatchItemFailures the
+      // handler returns only the failed record's sequence number, so just that
+      // record retries + bisects toward the DLQ while its siblings commit.
+      reportBatchItemFailures: true,
+    }));
+
+    NagSuppressions.addResourceSuppressions(this.fn, [
+      {
+        id: 'AwsSolutions-IAM4',
+        reason: 'AWSLambdaBasicExecutionRole is required for CloudWatch Logs access',
+      },
+      {
+        id: 'AwsSolutions-IAM5',
+        reason:
+          'DynamoDB index/* + stream ARN wildcards generated by CDK grantReadWriteData '
+          + '(ChildTaskIndex query) and the DynamoEventSource read access',
+      },
+    ], true);
+
+    NagSuppressions.addResourceSuppressions(this.dlq, [
+      {
+        id: 'AwsSolutions-SQS3',
+        reason:
+          'This queue IS the DLQ for the reconciler stream consumer — having its own DLQ would be infinite recursion',
+      },
+    ]);
+  }
+}

cdk/src/constructs/stranded-orchestration-reconciler.ts

@@ -0,0 +1,126 @@
+/**
+ *  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 * as path from 'path';
+import { Duration } from 'aws-cdk-lib';
+import * as dynamodb from 'aws-cdk-lib/aws-dynamodb';
+import * as events from 'aws-cdk-lib/aws-events';
+import * as targets from 'aws-cdk-lib/aws-events-targets';
+import { Architecture, Runtime } from 'aws-cdk-lib/aws-lambda';
+import * as lambda from 'aws-cdk-lib/aws-lambda-nodejs';
+import { NagSuppressions } from 'cdk-nag';
+import { Construct } from 'constructs';
+
+/**
+ * Properties for StrandedOrchestrationReconciler construct.
+ */
+export interface StrandedOrchestrationReconcilerProps {
+  /** OrchestrationTable — read DAG state, write recovered child statuses. */
+  readonly orchestrationTable: dynamodb.ITable;
+  /** TaskTable — read released children's task status (terminal? built?) + createTaskCore writes. */
+  readonly taskTable: dynamodb.ITable;
+  /** TaskEventsTable — createTaskCore writes task_created events. */
+  readonly taskEventsTable: dynamodb.ITable;
+  /** Orchestrator function ARN — releaseChild → createTaskCore async-invokes it. */
+  readonly orchestratorFunctionArn?: string;
+  /**
+   * Sweep cadence. Long enough to amortise the scan; short enough to
+   * clear a lost-event stall in a reasonable user-facing time.
+   * @default Duration.minutes(10)
+   */
+  readonly schedule?: Duration;
+}
+
+/**
+ * Scheduled backstop for sub-issue orchestration.
+ *
+ * The live ``OrchestrationReconciler`` reacts to TaskTable-stream terminal
+ * events to release dependency-unblocked children. If it is unavailable
+ * when an event fires (deploy/throttle/OOM/DLQ-parked record) that event
+ * is lost and the orchestration stalls. This scheduled sweep re-derives
+ * gating truth from persisted state and recovers stranded children
+ * (see ``handlers/reconcile-stranded-orchestrations.ts``).
+ *
+ * Mirrors ``StrandedTaskReconciler``. Grants match the live reconciler
+ * because it runs the same ``createTaskCore`` release path in-process.
+ */
+
+/** Sweep Lambda timeout (minutes) — matches the live reconciler's createTaskCore
+ *  + Bedrock/S3 SDK bundle cold-start + release work. */
+const SWEEP_TIMEOUT_MINUTES = 5;
+
+export class StrandedOrchestrationReconciler extends Construct {
+  public readonly fn: lambda.NodejsFunction;
+
+  constructor(scope: Construct, id: string, props: StrandedOrchestrationReconcilerProps) {
+    super(scope, id);
+
+    const handlersDir = path.join(__dirname, '..', 'handlers');
+
+    this.fn = new lambda.NodejsFunction(this, 'ReconcilerFn', {
+      entry: path.join(handlersDir, 'reconcile-stranded-orchestrations.ts'),
+      handler: 'handler',
+      runtime: Runtime.NODEJS_24_X,
+      architecture: Architecture.ARM_64,
+      timeout: Duration.minutes(SWEEP_TIMEOUT_MINUTES),
+      // 512 MB to match the live reconciler — same createTaskCore +
+      // Bedrock/S3 SDK bundle (see OrchestrationReconciler memory note).
+      memorySize: 512,
+      environment: {
+        ORCHESTRATION_TABLE_NAME: props.orchestrationTable.tableName,
+        TASK_TABLE_NAME: props.taskTable.tableName,
+        TASK_EVENTS_TABLE_NAME: props.taskEventsTable.tableName,
+        ...(props.orchestratorFunctionArn && {
+          ORCHESTRATOR_FUNCTION_ARN: props.orchestratorFunctionArn,
+        }),
+      },
+      bundling: {
+        externalModules: ['@aws-sdk/*'],
+        // Imports createTaskCore (releases children, which can carry attachments) —
+        // its transitive attachment-screening PDF path needs pdf-parse unbundled.
+        // Kept in lockstep with the other attachment-screening constructs; see the
+        // //:check:pdf-parse-bundling guard that enforces this.
+        nodeModules: ['pdf-parse'],
+      },
+    });
+
+    props.orchestrationTable.grantReadWriteData(this.fn);
+    props.taskTable.grantReadWriteData(this.fn);
+    props.taskEventsTable.grantReadWriteData(this.fn);
+
+    const schedule = props.schedule ?? Duration.minutes(10);
+    const rule = new events.Rule(this, 'SweepSchedule', {
+      schedule: events.Schedule.rate(schedule),
+    });
+    rule.addTarget(new targets.LambdaFunction(this.fn));
+
+    NagSuppressions.addResourceSuppressions(this.fn, [
+      {
+        id: 'AwsSolutions-IAM4',
+        reason: 'AWSLambdaBasicExecutionRole is required for CloudWatch Logs access',
+      },
+      {
+        id: 'AwsSolutions-IAM5',
+        reason:
+          'DynamoDB index/* wildcards generated by CDK grantReadWriteData for the '
+          + 'orchestration scan + child-task lookups + createTaskCore write path',
+      },
+    ], true);
+  }
+}

cdk/src/constructs/task-table.ts

@@ -57,8 +57,11 @@ export interface TaskTableProps {
  * - UserStatusIndex (PK: user_id, SK: status_created_at) — "my tasks" queries
  * - StatusIndex (PK: status, SK: created_at) — queue processing, monitoring
  * - IdempotencyIndex (PK: idempotency_key) — sparse index for dedup
+ * - LinearIssueIndex (PK: linear_issue_id, SK: created_at) — sparse; resolve a
+ *   Linear issue back to its newest ABCA task + PR (the standalone
+ *   comment trigger)
  * - JiraIssueIndex (PK: jira_issue_identity, SK: created_at) — sparse index
- *   for resolving a Jira issue to its newest PR-producing task
+ *   for resolving a Jira issue to its newest PR-producing task (#640)
  */
 export class TaskTable extends Construct {
   /**
@@ -79,6 +82,16 @@ export class TaskTable extends Construct {
    */
   public static readonly IDEMPOTENCY_INDEX = 'IdempotencyIndex';
 
+  /**
+   * GSI name for resolving a Linear issue → its newest ABCA task + PR.
+   * PK: linear_issue_id, SK: created_at (newest task wins). Sparse —
+   * only Linear-origin tasks (which write the top-level ``linear_issue_id``
+   * attribute) are projected; GitHub/Slack/API tasks are absent. Powers the
+   * standalone ``@bgagent`` comment trigger on a plain (non-orchestration)
+   * issue, where no orchestration row records the issue→PR link.
+   */
+  public static readonly LINEAR_ISSUE_INDEX = 'LinearIssueIndex';
+
   /**
    * GSI for resolving a tenant-scoped Jira issue to its newest ABCA task.
    * PK: jira_issue_identity (`{cloudId}#{issueKey}`), SK: created_at.
@@ -104,6 +117,17 @@ export class TaskTable extends Construct {
       pointInTimeRecoverySpecification: {
         pointInTimeRecoveryEnabled: props.pointInTimeRecovery ?? true,
       },
+      // NEW_IMAGE stream feeds the orchestration reconciler
+      // (`OrchestrationReconciler`), which reacts to child tasks reaching
+      // terminal status to release dependency-unblocked children. This is
+      // the table's FIRST and only stream consumer — deliberately on
+      // TaskTable rather than TaskEventsTable, whose stream is already at
+      // its 2-consumer limit (FanOutConsumer + ApprovalMetricsPublisher;
+      // see TaskEventsTable). NEW_IMAGE suffices — the reconciler reads
+      // status/build_passed/orchestration_id off the new record image.
+      // Enabling a stream on an existing table is an in-place CFN update
+      // (no table replacement).
+      stream: dynamodb.StreamViewType.NEW_IMAGE,
       removalPolicy: props.removalPolicy ?? RemovalPolicy.DESTROY,
     });
 
@@ -130,9 +154,28 @@ export class TaskTable extends Construct {
       projectionType: dynamodb.ProjectionType.KEYS_ONLY,
     });
 
-    // Sparse: only Jira-origin tasks with issue metadata carry the top-level
-    // jira_issue_identity attribute. Keep the projection limited to fields the
-    // comment-trigger resolver needs.
+    // GSI: Linear issue → newest ABCA task + PR (sparse — only Linear-origin
+    // tasks carry the top-level linear_issue_id), for the standalone comment
+    // trigger. INCLUDE-projects just the fields the trigger reads, so
+    // the index stays lean (no full-item copy on every task write).
+    this.table.addGlobalSecondaryIndex({
+      indexName: TaskTable.LINEAR_ISSUE_INDEX,
+      partitionKey: { name: 'linear_issue_id', type: dynamodb.AttributeType.STRING },
+      sortKey: { name: 'created_at', type: dynamodb.AttributeType.STRING },
+      projectionType: dynamodb.ProjectionType.INCLUDE,
+      // NOTE: a GSI's projection CANNOT be changed in place — DynamoDB rejects it
+      // ("Cannot update GSI's properties other than Provisioned Throughput…";
+      // observed when deploying the iteration reply). So the iteration reply's
+      // running-total query does a per-task GetItem for cost_usd rather than
+      // widening this projection. Keep this list as-is unless you create a NEW
+      // index with a different name.
+      nonKeyAttributes: ['pr_url', 'pr_number', 'status', 'repo', 'user_id', 'channel_metadata'],
+    });
+
+    // GSI: Jira issue → newest PR-producing task (#640 comment-triggered PR
+    // iteration). Sparse: only Jira-origin tasks with issue metadata carry the
+    // top-level jira_issue_identity attribute. Same lean INCLUDE projection as
+    // the Linear index above — just the fields the comment-trigger resolver reads.
     this.table.addGlobalSecondaryIndex({
       indexName: TaskTable.JIRA_ISSUE_INDEX,
       partitionKey: { name: 'jira_issue_identity', type: dynamodb.AttributeType.STRING },