fix: prevent tasks from being associated with the wrong PR

apps/api/src/services/git-platform/github.test.ts

@@ -50,7 +50,7 @@ describe("GitHubPlatform", () => {
         merged: false,
         mergeable: true,
         draft: false,
-        head: { sha: "abc123" },
+        head: { sha: "abc123", ref: "optio/task-abc" },
         base: { ref: "main" },
         html_url: "https://github.com/acme/widgets/pull/42",
         user: { login: "alice" },
@@ -74,6 +74,7 @@ describe("GitHubPlatform", () => {
       expect(pr.number).toBe(42);
       expect(pr.title).toBe("Fix bug");
       expect(pr.headSha).toBe("abc123");
+      expect(pr.headBranch).toBe("optio/task-abc");
       expect(pr.merged).toBe(false);
       expect(pr.labels).toEqual(["bug"]);
     });

apps/api/src/services/git-platform/github.ts

@@ -260,6 +260,7 @@ function mapPr(data: any): PullRequest {
     mergeable: data.mergeable ?? null,
     draft: data.draft ?? false,
     headSha: data.head?.sha ?? "",
+    headBranch: data.head?.ref ?? "",
     baseBranch: data.base?.ref ?? "",
     url: data.html_url ?? "",
     author: data.user?.login ?? "",

apps/api/src/services/git-platform/gitlab.test.ts

@@ -51,6 +51,7 @@ describe("GitLabPlatform", () => {
         draft: false,
         sha: "def456",
         diff_refs: { head_sha: "def456" },
+        source_branch: "optio/task-abc",
         target_branch: "main",
         web_url: "https://gitlab.com/acme/widgets/-/merge_requests/7",
         author: { username: "alice" },
@@ -78,6 +79,7 @@ describe("GitLabPlatform", () => {
       expect(pr.merged).toBe(false);
       expect(pr.mergeable).toBe(true);
       expect(pr.headSha).toBe("def456");
+      expect(pr.headBranch).toBe("optio/task-abc");
     });
 
     it("maps merged MR state correctly", async () => {

apps/api/src/services/git-platform/gitlab.ts

@@ -396,6 +396,7 @@ function mapMr(data: any, ri: RepoIdentifier): PullRequest {
           : null,
     draft: data.draft ?? data.work_in_progress ?? false,
     headSha: data.sha ?? data.diff_refs?.head_sha ?? "",
+    headBranch: data.source_branch ?? "",
     baseBranch: data.target_branch ?? "",
     url: data.web_url ?? `https://${ri.host}/${ri.owner}/${ri.repo}/-/merge_requests/${data.iid}`,
     author: data.author?.username ?? "",

apps/api/src/services/pr-detection-service.test.ts

@@ -1,10 +1,11 @@
 import { describe, it, expect, vi, beforeEach } from "vitest";
-import { parseOwnerRepo, checkExistingPr } from "./pr-detection-service.js";
+import { parseOwnerRepo, checkExistingPr, validateTaskPrUrl } from "./pr-detection-service.js";
 
 // Mock git-token-service
 const mockPlatform = {
   type: "github",
   listOpenPullRequests: vi.fn(),
+  getPullRequest: vi.fn(),
 };
 const mockGetGitPlatformForRepo = vi.fn();
 
@@ -87,6 +88,7 @@ describe("checkExistingPr", () => {
         mergeable: true,
         draft: false,
         headSha: "abc",
+        headBranch: "optio/task-123",
         baseBranch: "main",
         author: "",
         assignees: [],
@@ -170,3 +172,69 @@ describe("checkExistingPr", () => {
     });
   });
 });
+
+describe("validateTaskPrUrl", () => {
+  beforeEach(() => {
+    vi.clearAllMocks();
+    mockGetGitPlatformForRepo.mockResolvedValue({
+      platform: mockPlatform,
+      ri: {
+        platform: "github",
+        host: "github.com",
+        owner: "owner",
+        repo: "repo",
+        apiBaseUrl: "https://api.github.com",
+      },
+    });
+  });
+
+  it("returns valid when the PR head branch is the task branch", async () => {
+    mockPlatform.getPullRequest.mockResolvedValue({ headBranch: "optio/task-abc" });
+    const result = await validateTaskPrUrl(
+      "https://github.com/owner/repo",
+      "abc",
+      "https://github.com/owner/repo/pull/506",
+    );
+    expect(result).toBe("valid");
+    expect(mockPlatform.getPullRequest).toHaveBeenCalledWith(expect.anything(), 506);
+  });
+
+  it("returns invalid when the PR head branch is a different branch", async () => {
+    mockPlatform.getPullRequest.mockResolvedValue({ headBranch: "renovate/aiosqlite-0.x" });
+    const result = await validateTaskPrUrl(
+      "https://github.com/owner/repo",
+      "abc",
+      "https://github.com/owner/repo/pull/453",
+    );
+    expect(result).toBe("invalid");
+  });
+
+  it("returns invalid when the URL is not a parseable PR URL", async () => {
+    const result = await validateTaskPrUrl(
+      "https://github.com/owner/repo",
+      "abc",
+      "https://github.com/owner/repo/pulls",
+    );
+    expect(result).toBe("invalid");
+  });
+
+  it("returns unknown when no git token is available", async () => {
+    mockGetGitPlatformForRepo.mockRejectedValue(new Error("no token"));
+    const result = await validateTaskPrUrl(
+      "https://github.com/owner/repo",
+      "abc",
+      "https://github.com/owner/repo/pull/506",
+    );
+    expect(result).toBe("unknown");
+  });
+
+  it("returns unknown when the PR fetch fails", async () => {
+    mockPlatform.getPullRequest.mockRejectedValue(new Error("GitHub API error 500"));
+    const result = await validateTaskPrUrl(
+      "https://github.com/owner/repo",
+      "abc",
+      "https://github.com/owner/repo/pull/506",
+    );
+    expect(result).toBe("unknown");
+  });
+});

apps/api/src/services/pr-detection-service.ts

@@ -1,4 +1,4 @@
-import { TASK_BRANCH_PREFIX, parseRepoUrl } from "@optio/shared";
+import { TASK_BRANCH_PREFIX, parseRepoUrl, parsePrUrl } from "@optio/shared";
 import { getGitPlatformForRepo } from "./git-token-service.js";
 import { logger } from "../logger.js";
 
@@ -64,3 +64,41 @@ export async function checkExistingPr(
     return null;
   }
 }
+
+export type PrUrlValidation = "valid" | "invalid" | "unknown";
+
+/**
+ * Check that a candidate PR URL (scraped from agent logs) actually points at
+ * this task's PR by comparing the PR's head branch to the deterministic task
+ * branch. "unknown" means the API could not be consulted — callers should
+ * accept the candidate in that case rather than block PR detection.
+ *
+ * Assumes `prUrl` has already been confirmed to point at `repoUrl`; only the
+ * PR number is taken from it, its owner/repo/host are not re-checked here.
+ */
+export async function validateTaskPrUrl(
+  repoUrl: string,
+  taskId: string,
+  prUrl: string,
+): Promise<PrUrlValidation> {
+  const parsed = parsePrUrl(prUrl);
+  if (!parsed) return "invalid";
+
+  let platform;
+  let ri;
+  try {
+    const result = await getGitPlatformForRepo(repoUrl, { server: true });
+    platform = result.platform;
+    ri = result.ri;
+  } catch {
+    return "unknown";
+  }
+
+  try {
+    const pr = await platform.getPullRequest(ri, parsed.prNumber);
+    return pr.headBranch === `${TASK_BRANCH_PREFIX}${taskId}` ? "valid" : "invalid";
+  } catch (err) {
+    logger.debug({ err, prUrl }, "Could not fetch candidate PR for validation");
+    return "unknown";
+  }
+}

apps/api/src/services/pr-review-service.test.ts

@@ -163,6 +163,7 @@ function setupPlatformMocks(
     title: "Add feature X",
     body: "Implements feature X",
     headSha: "abc123",
+    headBranch: "optio/task-42",
     number: 42,
     state: "open" as const,
     merged: false,
@@ -283,6 +284,7 @@ describe("launchPrReview", () => {
       title: "Fix bug",
       body: "Fixes bug",
       headSha: "def456",
+      headBranch: "optio/task-10",
       number: 10,
       state: "open",
       merged: false,
@@ -331,6 +333,7 @@ describe("launchPrReview", () => {
       title: "Chore",
       body: "",
       headSha: "sha1",
+      headBranch: "optio/task-5",
       number: 5,
       state: "open",
       merged: false,
@@ -375,6 +378,7 @@ describe("launchPrReview", () => {
       title: "PR",
       body: "",
       headSha: "",
+      headBranch: "optio/task-99",
       number: 99,
       state: "open",
       merged: false,

apps/api/src/services/reconcile-executor.test.ts

@@ -113,6 +113,7 @@ function repoSnapshot(overrides: Partial<WorldSnapshot> = {}): WorldSnapshot {
       blocksParent: false,
       workspaceId: "ws-1",
       workflowRunId: null,
+      createdAt: new Date("2026-01-01T00:00:00Z"),
     },
     status: {
       state: TaskState.QUEUED,

apps/api/src/services/reconcile-snapshot.ts

@@ -180,14 +180,15 @@ function loadRepoRun(row: typeof tasks.$inferSelect, ref: RunRef): Run {
     agentType: row.agentType,
     prompt: row.prompt,
     title: row.title,
-    taskType: (row.taskType as "coding" | "review") ?? "coding",
+    taskType: (row.taskType as "coding" | "review" | "pr_review") ?? "coding",
     maxRetries: row.maxRetries,
     priority: row.priority,
     ignoreOffPeak: row.ignoreOffPeak,
     parentTaskId: row.parentTaskId ?? null,
     blocksParent: row.blocksParent,
     workspaceId: row.workspaceId ?? null,
     workflowRunId: row.workflowRunId ?? null,
+    createdAt: row.createdAt,
   };
   const status: RepoRunStatus = {
     state: row.state as TaskState,
@@ -305,6 +306,7 @@ async function loadPrStatus(run: Run, userId: string | null): Promise<PrStatus |
     checksStatus,
     reviewStatus: reviewResult.status as PrStatus["reviewStatus"],
     latestReviewComments: reviewResult.comments || null,
+    createdAt: prData.createdAt || null,
   };
 }
 

apps/api/src/services/repo-pool-service.ts

@@ -1386,6 +1386,27 @@ export async function deleteEnvoyConfigMap(podName: string): Promise<void> {
  *
  * Returns true if processes were found and killed, false otherwise.
  */
+/**
+ * Check whether any process belonging to a task is still running in its pod.
+ * Used after an exec stream ends: the transport can tear down with a status
+ * frame that falsely reads as a clean exit while the agent lives on.
+ */
+export async function isTaskAgentRunning(podId: string, taskId: string): Promise<boolean> {
+  const [pod] = await db.select().from(repoPods).where(eq(repoPods.id, podId));
+  if (!pod?.podName) return false;
+  const rt = getRuntime();
+  const handle: ContainerHandle = { id: pod.podId ?? pod.podName, name: pod.podName };
+
+  const checkScript = `grep -rl "OPTIO_TASK_ID=${taskId}" /proc/*/environ 2>/dev/null | head -1`;
+  const session = await rt.exec(handle, ["bash", "-c", checkScript], { tty: false });
+  let output = "";
+  for await (const chunk of session.stdout as AsyncIterable<Buffer>) {
+    output += chunk.toString();
+  }
+  session.close();
+  return output.trim().length > 0;
+}
+
 export async function killOrphanedAgentInPod(podId: string, taskId: string): Promise<boolean> {
   const [pod] = await db.select().from(repoPods).where(eq(repoPods.id, podId));
   if (!pod || !pod.podName || pod.state !== "ready") return false;

apps/api/src/services/task-service.ts

@@ -429,6 +429,13 @@ export async function updateTaskPr(id: string, prUrl: string) {
     .where(eq(tasks.id, id));
 }
 
+export async function incrementTaskRetryCount(id: string) {
+  await db
+    .update(tasks)
+    .set({ retryCount: sql`${tasks.retryCount} + 1`, updatedAt: new Date() })
+    .where(eq(tasks.id, id));
+}
+
 export async function updateTaskSession(id: string, sessionId: string) {
   await db.update(tasks).set({ sessionId, updatedAt: new Date() }).where(eq(tasks.id, id));
 }

apps/api/src/services/workflow-pool-service.ts

@@ -455,6 +455,60 @@ export async function execRunInPod(
   return rt.exec(handle, ["bash", "-c", script], { tty: false });
 }
 
+/**
+ * Check whether any process belonging to a workflow run is still running in
+ * its pod. Used after an exec stream ends: the transport can tear down with
+ * a status frame that falsely reads as a clean exit while the agent lives on.
+ */
+export async function isRunAgentRunning(pod: WorkflowPod, runId: string): Promise<boolean> {
+  const rt = getRuntime();
+  const handle: ContainerHandle = { id: pod.podId ?? pod.podName!, name: pod.podName! };
+
+  const checkScript = `grep -rl "OPTIO_WORKFLOW_RUN_ID=${runId}" /proc/*/environ 2>/dev/null | head -1`;
+  const session = await rt.exec(handle, ["bash", "-c", checkScript], { tty: false });
+  let output = "";
+  for await (const chunk of session.stdout as AsyncIterable<Buffer>) {
+    output += chunk.toString();
+  }
+  session.close();
+  return output.trim().length > 0;
+}
+
+/**
+ * Kill any agent processes belonging to a workflow run that outlived its
+ * exec stream (e.g. the connection was severed mid-run). Workflow pods are
+ * shared across runs, so a zombie agent would oversubscribe the pod and can
+ * produce duplicate external side effects when the run is retried.
+ */
+export async function killOrphanedRunInPod(pod: WorkflowPod, runId: string): Promise<boolean> {
+  const rt = getRuntime();
+  const handle: ContainerHandle = { id: pod.podId ?? pod.podName!, name: pod.podName! };
+
+  const killScript = [
+    `pids=$(grep -rl "OPTIO_WORKFLOW_RUN_ID=${runId}" /proc/*/environ 2>/dev/null | cut -d/ -f3 | sort -u || true)`,
+    `if [ -n "$pids" ]; then`,
+    `  kill -TERM $pids 2>/dev/null || true`,
+    `  sleep 2`,
+    `  kill -9 $pids 2>/dev/null || true`,
+    `  echo "killed"`,
+    `else`,
+    `  echo "none"`,
+    `fi`,
+  ].join("\n");
+
+  const killSession = await rt.exec(handle, ["bash", "-c", killScript], { tty: false });
+  let output = "";
+  for await (const chunk of killSession.stdout as AsyncIterable<Buffer>) {
+    output += chunk.toString();
+  }
+  killSession.close();
+  const killed = output.includes("killed");
+  if (killed) {
+    logger.info({ podName: pod.podName, runId }, "Killed orphaned workflow agent processes");
+  }
+  return killed;
+}
+
 /**
  * Decrement the active run count for a workflow pod. Clamped at zero so a
  * double-release (e.g. zombie cleanup + worker finally) can't drive it negative.