feat(jira): mature iteration feedback in place (#697)

cdk/src/constructs/iteration-heartbeat.ts

@@ -52,14 +52,13 @@ export interface IterationHeartbeatProps {
  * Mid-run liveness heartbeat (scheduled).
  *
  * A scheduled Lambda that finds RUNNING comment-triggered iteration tasks and
- * EDITS the existing maturing Linear reply in place to show liveness ("🔄
+ * EDITS the existing maturing Linear/Jira comment in place to show liveness ("🔄
  * Working — updating PR #N… _8m elapsed_"), so a long run isn't a silent black
  * box between 👀 and the terminal ✅/❌ (observed in practice as a 22-min silence).
  *
- * The construct owns only the Lambda + schedule + TaskTable read. The Linear
- * workspace-registry env + per-workspace OAuth ``GetSecretValue`` grant are
- * wired by the stack after instantiation (mirrors OrchestrationReconciler),
- * since they belong to the LinearIntegration construct.
+ * The construct owns only the Lambda + schedule + TaskTable read. Surface
+ * registry environments and OAuth grants are wired by the stack after
+ * instantiation, since they belong to their integration constructs.
  */
 export class IterationHeartbeat extends Construct {
   public readonly fn: lambda.NodejsFunction;
@@ -86,7 +85,7 @@ export class IterationHeartbeat extends Construct {
     });
 
     // Read-only on the TaskTable (StatusIndex query). No write — a heartbeat
-    // never mutates task state; it only edits a Linear comment.
+    // never mutates task state; it only edits a surface comment.
     props.taskTable.grantReadData(this.fn);
 
     const schedule = props.schedule ?? Duration.minutes(DEFAULT_SCHEDULE_MINUTES);
@@ -103,7 +102,7 @@ export class IterationHeartbeat extends Construct {
       {
         id: 'AwsSolutions-IAM5',
         reason: 'DynamoDB index/* wildcard generated by CDK grantReadData; '
-          + 'per-workspace linear-oauth secret prefix grant added by the stack',
+          + 'scoped per-surface OAuth secret prefix grants added by the stack',
       },
     ], true);
   }

cdk/src/handlers/fanout-task-events.ts

@@ -55,9 +55,9 @@ import { claimTerminalReply, releaseReplyClaim, terminalReplyClaimed } from './s
 import {
   buildAdfDocument,
   postIssueCommentAdf,
-  type AdfParagraph,
-  type AdfTextRun,
+  updateIssueCommentAdf,
 } from './shared/jira-feedback';
+import { renderJiraFinalStatusComment } from './shared/jira-status-comment';
 import { EMOJI_FAILURE, EMOJI_NEEDS_INPUT, EMOJI_SUCCESS, postIssueComment, swapCommentReaction, upsertThreadedReply } from './shared/linear-feedback';
 import { logger } from './shared/logger';
 import { coerceNumericOrNull } from './shared/numeric';
@@ -206,10 +206,11 @@ export const CHANNEL_DEFAULTS: Record<NotificationChannel, ReadonlySet<string>>
   // now emits it as a distinct terminal event (``orchestrator.ts``); the
   // Slack + email defaults already subscribe to it.
   //
-  // Jira has no comment-edit API (same as Linear), so this is post-once:
-  // idempotency across partial-batch retries rides on the
-  // ``jira_final_comment_event_id`` marker. The agent-side start comment
-  // ("🤖 ABCA picked up this issue…") stays for in-flight progress.
+  // Ordinary Jira tasks receive a post-once terminal comment, with idempotency
+  // carried by ``jira_final_comment_event_id``. Comment-triggered iterations
+  // instead edit their stored status comment under a terminal-writer claim.
+  // The agent-side start comment ("🤖 ABCA picked up this issue…") stays for
+  // ordinary tasks' in-flight progress.
   jira: new Set<string>([
     ...TERMINAL_EVENT_TYPES,
     'task_timed_out',
@@ -691,10 +692,9 @@ async function saveLinearPrCommentState(taskId: string, eventId: string): Promis
 }
 
 /**
- * Persist the post-once marker after a successful Jira final-status comment
- * (see ``dispatchToJira``). The Jira analogue of ``saveLinearCommentState`` —
- * Jira has no comment-edit API, so the marker is what makes the post
- * idempotent across partial-batch retries.
+ * Persist the post-once marker after a successful ordinary Jira final-status
+ * comment (see ``dispatchToJira``). Iteration comments are edited in place and
+ * use the shared terminal-reply claim instead.
  */
 async function saveJiraCommentState(taskId: string, eventId: string): Promise<void> {
   await saveDispatchMarker({
@@ -1546,93 +1546,7 @@ async function sumIterationCostForIssue(
   });
 }
 
-/**
- * Render the Jira final-status comment as ADF paragraphs. Mirrors
- * ``renderLinearFinalStatusComment`` framing — the difference is the output
- * shape (ADF runs vs Markdown string), because Jira REST v3 comments require
- * Atlassian Document Format, not Markdown.
- *
- * Three outcomes based on ``(eventType, prUrl)``:
- *
- *   1. ``task_completed``                        → ✅ "Task completed"
- *   2. any non-completed terminal event WITH PR  → ⚠️ "Shipped a PR but stopped early"
- *   3. any non-completed terminal event NO PR    → ❌ "Task <subtype>" + classifier title
- *
- * The PR URL is rendered on the ✅ success path too — not just the ⚠️ path —
- * because the agent's own "PR opened" comment is not guaranteed to have fired
- * (an agent that skipped that step), so the platform comment must always carry
- * the link or it can be lost entirely.
- * ``renderLinearFinalStatusComment`` does the same for Linear.
- *
- * Missing metric values render as ``—``. The result is a list of ADF
- * paragraphs (blank lines are empty paragraphs — ADF text nodes do not
- * honor ``\n``), fed to ``buildAdfDocument``.
- */
-export function renderJiraFinalStatusComment(args: {
-  eventType: string;
-  prUrl: string | null;
-  costUsd: number | null;
-  turns: number | null;
-  maxTurns: number | null;
-  durationS: number | null;
-  taskId: string;
-  errorTitle: string | null;
-}): ReadonlyArray<AdfParagraph> {
-  const isCompleted = args.eventType === 'task_completed';
-  const shippedDespiteFailure = !isCompleted && args.prUrl != null;
-
-  // Header runs. Bold scope mirrors Linear's Markdown: the ⚠️ frame bolds
-  // only through the reason and leaves the trailing "review and decide…"
-  // advice unbolded, so it's a two-run paragraph. The ✅ / ❌ frames are a
-  // single bold run.
-  let headerRuns: AdfTextRun[];
-  if (isCompleted) {
-    headerRuns = [{ text: '✅ Task completed', strong: true }];
-  } else if (shippedDespiteFailure) {
-    const reason = args.errorTitle ? ` — ${args.errorTitle}` : '';
-    headerRuns = [
-      { text: `⚠️ Shipped a PR but stopped early${reason}`, strong: true },
-      { text: ' — review and decide if more work is needed' },
-    ];
-  } else {
-    // Humanize the event subtype for the header: strip the ``task_`` prefix
-    // and turn underscores into spaces so ``task_timed_out`` reads "Task
-    // timed out" rather than the raw "Task timed_out". Jira is the only
-    // channel routing ``task_timed_out`` through this renderer, so this
-    // multi-word subtype is a case the copied-from-Linear code never hit.
-    const subtype = args.eventType.replace(/^task_/, '').replace(/_/g, ' ');
-    const reason = args.errorTitle ? `: ${args.errorTitle}` : '';
-    headerRuns = [{ text: `❌ Task ${subtype}${reason}`, strong: true }];
-  }
-
-  const costStr = args.costUsd != null ? `$${args.costUsd.toFixed(2)}` : '—';
-  const turnsStr = args.turns != null
-    ? `${args.turns}${args.maxTurns != null ? ` / ${args.maxTurns}` : ''}`
-    : '—';
-  const durationStr = args.durationS != null
-    ? formatDuration(args.durationS)
-    : '—';
-
-  const paragraphs: AdfParagraph[] = [
-    headerRuns,
-    [{ text: `cost: ${costStr} • turns: ${turnsStr} • duration: ${durationStr}` }],
-  ];
-  // Render the PR link whenever one exists — on both the ✅ success path and
-  // the ⚠️ shipped-but-stopped path — because the agent's own "PR opened"
-  // comment is not guaranteed to have fired, so this is the only guaranteed
-  // PR-link surface. The URL run carries an ``href`` so it renders as a
-  // clickable hyperlink — ADF does not auto-linkify a bare URL in a plain
-  // text node the way Linear's Markdown does, so without this the requester
-  // would have to copy-paste it.
-  if (args.prUrl) {
-    paragraphs.push([
-      { text: 'PR: ' },
-      { text: args.prUrl, href: args.prUrl },
-    ]);
-  }
-  paragraphs.push([{ text: `task ${args.taskId}`, em: true }]);
-  return paragraphs;
-}
+export { renderJiraFinalStatusComment };
 
 /**
  * Jira dispatcher — posts a deterministic final-status comment when a
@@ -1697,11 +1611,18 @@ async function dispatchToJira(event: FanOutEvent): Promise<void> {
     return;
   }
 
-  // Idempotency across partial-batch retries: Jira has no comment edit API,
-  // so a re-run (e.g. a sibling channel's infra rejection pushed the whole
-  // stream record into batchItemFailures) would post a duplicate. The
-  // marker is persisted after the first successful post below.
-  if (task.jira_final_comment_event_id) {
+  const iterationReplyId = task.channel_metadata?.iteration_reply_comment_id;
+  const isIteration = Boolean(task.channel_metadata?.trigger_comment_id);
+  const isOrchestratedIteration =
+    task.channel_metadata?.orchestration_iteration === 'true';
+  // The reconciler owns an orchestrated iteration's terminal comment because it
+  // must settle that comment before continuing the dependent restack cascade.
+  if (isOrchestratedIteration) return;
+
+  // Ordinary Jira terminal comments are created once. A re-run (for example,
+  // because a sibling channel rejected the record) would otherwise duplicate
+  // that comment, so persist a marker after the first successful post below.
+  if (!isIteration && task.jira_final_comment_event_id) {
     logger.info('[fanout/jira] final comment already posted — skipping (idempotent retry)', {
       event: 'fanout.jira.already_posted',
       task_id: task.task_id,
@@ -1746,6 +1667,55 @@ async function dispatchToJira(event: FanOutEvent): Promise<void> {
     errorTitle: classification?.title ?? null,
   });
 
+  if (isIteration && iterationReplyId) {
+    const tableName = process.env.TASK_TABLE_NAME;
+    if (!tableName) return;
+    const claim = await claimTerminalReply(
+      ddb,
+      tableName,
+      task.task_id,
+      event.timestamp,
+    );
+    if (!claim.won) return;
+
+    const updateResult = await updateIssueCommentAdf(
+      { cloudId, registryTableName },
+      issueKey,
+      iterationReplyId,
+      buildAdfDocument(paragraphs),
+    );
+    if (updateResult.ok) {
+      logger.info('[fanout/jira] iteration comment matured in place', {
+        event: 'fanout.jira.iteration_matured',
+        task_id: task.task_id,
+        jira_issue_key: issueKey,
+        comment_id: iterationReplyId,
+      });
+      return;
+    }
+
+    const release = await releaseReplyClaim(
+      ddb,
+      tableName,
+      task.task_id,
+      claim.stamp,
+    );
+    logger.warn('[fanout/jira] iteration comment update failed (non-fatal)', {
+      event: 'fanout.jira.iteration_update_failed',
+      task_id: task.task_id,
+      jira_issue_key: issueKey,
+      comment_id: iterationReplyId,
+      retryable: updateResult.retryable,
+      release,
+    });
+    if (updateResult.retryable && release !== 'exhausted') {
+      throw new Error(
+        `[fanout/jira] transient Jira iteration update failure for task ${task.task_id}`,
+      );
+    }
+    return;
+  }
+
   const postResult = await postIssueCommentAdf(
     { cloudId, registryTableName },
     issueKey,

cdk/src/handlers/iteration-heartbeat-sweep.ts

@@ -35,16 +35,22 @@
  * (reconciler) later overwrites the working line with ✅/❌ as today.
  */
 
-import { DynamoDBClient, QueryCommand } from '@aws-sdk/client-dynamodb';
+import { DynamoDBClient, GetItemCommand, QueryCommand } from '@aws-sdk/client-dynamodb';
 import { planHeartbeat, type HeartbeatTaskView } from './shared/iteration-heartbeat';
 import { logger } from './shared/logger';
-import { makeLinearChannel } from './shared/orchestration-channel-linear';
+import {
+  channelForSource,
+  type ChannelRegistryTables,
+} from './shared/orchestration-channel-factory';
 import { makeClient } from './shared/ua';
 
 const ddb = makeClient(DynamoDBClient);
 const TASK_TABLE = process.env.TASK_TABLE_NAME!;
 const STATUS_INDEX = process.env.TASK_STATUS_INDEX_NAME ?? 'StatusIndex';
-const WORKSPACE_REGISTRY_TABLE = process.env.LINEAR_WORKSPACE_REGISTRY_TABLE_NAME;
+const CHANNEL_REGISTRY_TABLES: ChannelRegistryTables = {
+  linear: process.env.LINEAR_WORKSPACE_REGISTRY_TABLE_NAME,
+  jira: process.env.JIRA_WORKSPACE_REGISTRY_TABLE_NAME,
+};
 
 /** Hard cap on tasks edited per sweep — a backstop against an unexpected flood. */
 const MAX_EDITS_PER_SWEEP = 50;
@@ -61,22 +67,43 @@ function toView(img: DdbMap): HeartbeatTaskView {
     ...(img.created_at?.S !== undefined && { createdAt: img.created_at.S }),
     ...(img.channel_source?.S !== undefined && { channelSource: img.channel_source.S }),
     ...(cm.linear_workspace_id?.S !== undefined && { linearWorkspaceId: cm.linear_workspace_id.S }),
+    ...(cm.jira_cloud_id?.S !== undefined && { jiraCloudId: cm.jira_cloud_id.S }),
     ...(cm.iteration_reply_comment_id?.S !== undefined && { iterationReplyCommentId: cm.iteration_reply_comment_id.S }),
     ...(cm.trigger_comment_id?.S !== undefined && { triggerCommentId: cm.trigger_comment_id.S }),
     // The issue the reply lives on. The orchestration path stamps
     // ``trigger_comment_issue_id`` (the parent epic, for a routed comment);
     // the STANDALONE path stamps only ``linear_issue_id`` (the reply is on that
     // same issue). Fall back to it so standalone iterations get a heartbeat —
     // the reconciler's reply path uses the same precedence.
-    ...((cm.trigger_comment_issue_id?.S ?? cm.linear_issue_id?.S) !== undefined && {
-      triggerCommentIssueId: cm.trigger_comment_issue_id?.S ?? cm.linear_issue_id?.S,
+    ...((cm.trigger_comment_issue_id?.S ?? cm.linear_issue_id?.S ?? cm.jira_issue_key?.S) !== undefined && {
+      triggerCommentIssueId:
+        cm.trigger_comment_issue_id?.S ?? cm.linear_issue_id?.S ?? cm.jira_issue_key?.S,
     }),
     isIteration: cm.orchestration_iteration?.S === 'true',
     ...(prNumberRaw !== undefined && { prNumber: Number(prNumberRaw) }),
     ...(img.pr_url?.S !== undefined && { prUrl: img.pr_url.S }),
   };
 }
 
+/** Strongly-consistent terminal guard immediately before a cosmetic edit. */
+async function terminalReplyAlreadyClaimed(taskId: string): Promise<boolean> {
+  try {
+    const result = await ddb.send(new GetItemCommand({
+      TableName: TASK_TABLE,
+      Key: { task_id: { S: taskId } },
+      ProjectionExpression: 'ack_replied_at',
+      ConsistentRead: true,
+    }));
+    return Boolean(result.Item?.ack_replied_at?.S);
+  } catch (err) {
+    logger.warn('Heartbeat sweep: terminal-claim check failed (non-fatal)', {
+      task_id: taskId,
+      error: err instanceof Error ? err.message : String(err),
+    });
+    return false;
+  }
+}
+
 /** Query every RUNNING task via the StatusIndex GSI (paginated). */
 async function loadRunningTasks(): Promise<DdbMap[]> {
   const items: DdbMap[] = [];
@@ -102,8 +129,8 @@ async function loadRunningTasks(): Promise<DdbMap[]> {
  * never wedge or alarm).
  */
 export async function handler(): Promise<void> {
-  if (!WORKSPACE_REGISTRY_TABLE) {
-    logger.info('Heartbeat sweep skipped — no Linear workspace registry configured');
+  if (!CHANNEL_REGISTRY_TABLES.linear && !CHANNEL_REGISTRY_TABLES.jira) {
+    logger.info('Heartbeat sweep skipped — no channel credentials registry configured');
     return;
   }
 
@@ -129,29 +156,35 @@ export async function handler(): Promise<void> {
     edited_cap: MAX_EDITS_PER_SWEEP,
   });
 
-  // The reply edit goes through the surface-agnostic channel; only the surface
-  // this sweep reads from (a Linear iteration reply) picks the adapter.
-  const channel = makeLinearChannel(WORKSPACE_REGISTRY_TABLE);
-
   let edited = 0;
   for (const plan of plans.slice(0, MAX_EDITS_PER_SWEEP)) {
     try {
-      await channel.upsertThreadedReply?.(
-        { issueId: plan.issueId, credentialsRef: plan.linearWorkspaceId },
-        { commentId: plan.parentCommentId },
-        plan.body,
-        { commentId: plan.replyId },
-        {
-          // Keep any already-landed deploy-preview block (a heartbeat must never
-          // clobber the screenshot the webhook may have appended).
-          preservePreview: true,
-          // A liveness tick is the LEAST important writer of this reply: if the
-          // task has already settled, saying "working" again would un-settle it
-          // in the reader's eyes.
-          skipIfSettled: true,
-        },
-      );
-      edited += 1;
+      if (await terminalReplyAlreadyClaimed(plan.taskId)) continue;
+      const channel = channelForSource(plan.channelSource, CHANNEL_REGISTRY_TABLES);
+      if (!channel) continue;
+      const issue = { issueId: plan.issueId, credentialsRef: plan.credentialsRef };
+      const ref = plan.channelSource === 'linear' && plan.parentCommentId
+        ? await channel.upsertThreadedReply?.(
+          issue,
+          { commentId: plan.parentCommentId },
+          plan.body,
+          { commentId: plan.replyId },
+          {
+            // Keep any already-landed deploy-preview block (a heartbeat must never
+            // clobber the screenshot the webhook may have appended).
+            preservePreview: true,
+            // A liveness tick is the LEAST important writer of this reply: if the
+            // task has already settled, saying "working" again would un-settle it
+            // in the reader's eyes.
+            skipIfSettled: true,
+          },
+        )
+        : await channel.upsertComment(
+          issue,
+          plan.body,
+          { commentId: plan.replyId },
+        );
+      if (ref) edited += 1;
     } catch (err) {
       logger.warn('Heartbeat sweep: reply edit failed (non-fatal)', {
         task_id: plan.taskId,