feat(otdf-sdk-mgr,.claude): idempotent orchestrate run + --force flag (DSPX-3302)

Re-running `orchestrate run` after a partial failure now resumes from where
it left off: cells whose branch already has an open PR are skipped (reported
as OK with the existing URL), and only failed/pending cells are dispatched.

- Add check_existing_pr() helper (gh pr list --head <branch>)
- run_cell() checks for existing PR before creating worktree or launching subagent
- --force flag bypasses the idempotency check (re-run even with existing PR)
- Dry-run annotates already-done cells with [PR EXISTS: <url>]
- SKILL.md: update Step 1/2 + replace "When to use partial runs" with
  "Resumption and partial runs" covering fix-and-retry, staging, and --force

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

Author: dmihalcik-virtru

.claude/skills/feature-orchestrate/SKILL.md

@@ -26,7 +26,7 @@ Before dispatching, run a dry-run so the user can confirm the topology and pinni
 uv run otdf-sdk-mgr orchestrate run xtest/features/<name>.yaml --dry-run
 ```
 
-The output names each cell, its target repo path, the branch it'll work on, and the worktree path the orchestrator will create (or `action=push+draft-PR` for the `tests` cell). Each wave is a set of cells with no dependencies between them; cells in a later wave have at least one `depends_on` edge into an earlier wave.
+The output names each cell, its target repo path, the branch it'll work on, and the worktree path the orchestrator will create (or `action=push+draft-PR` for the `tests` cell). Cells that already have an open PR are annotated with `[PR EXISTS: <url>]` — those will be skipped on the real run (no subagent launched). Each wave is a set of cells with no dependencies between them; cells in a later wave have at least one `depends_on` edge into an earlier wave.
 
 Surface the dry-run output to the user verbatim. If anything looks wrong (a cell going to the wrong repo, a missing `depends_on` edge, a stale branch name), ask the user to fix the spec via `feature-design` (or edit it directly) before proceeding.
 
@@ -38,7 +38,9 @@ When the user confirms, run for real:
 uv run otdf-sdk-mgr orchestrate run xtest/features/<name>.yaml
 ```
 
-For each cell, the orchestrator:
+The run is **idempotent**: cells whose branch already has an open PR are reported as `OK` with the existing PR URL and no subagent is launched. Re-running after a partial failure resumes from where it left off — already-done cells are skipped, failed or not-yet-started cells are dispatched. Use `--force` to re-run a cell even if it already has a PR (e.g. to incorporate spec changes into an existing draft).
+
+For each cell that needs work, the orchestrator:
 
 1. Creates `~/Documents/GitHub/worktrees/<JIRA-KEY>-<cell-key>/` as a worktree of `~/Documents/GitHub/opentdf/<path>` on branch `<cell.branch>`. Idempotent — reuses an existing worktree if it's already on the right branch, bails if it's on a different one.
 2. Writes a minimal `.claude/settings.json` into the worktree (allowing `git`, `gh pr create`, and the repo-type-appropriate test commands: `go`/`make`/`buf` for platform, `mvn` for java-sdk, `npm` for web-sdk).
@@ -62,10 +64,18 @@ web-sdk                  FAIL   exit 1
 
 Pass the table on to the user, plus the JSONL transcript paths for any FAIL rows so they can inspect what went wrong.
 
-## When to use partial runs
+## Resumption and partial runs
+
+**Resuming after a failure** — just re-run the same command. The orchestrator skips cells that already have open PRs and only dispatches the ones that failed or were skipped due to upstream failures. The dependency check still runs: if a cell's `depends_on` failed in the previous wave, it will be skipped again rather than dispatched into a broken state. Fix the upstream cell first (see below), then re-run.
+
+**Fixing a failed cell** — inspect the transcript at `.claude/tmp/runs/<JIRA-KEY>-<cell-key>.jsonl`, identify the problem, fix it (in the worktree at `~/Documents/GitHub/worktrees/<JIRA-KEY>-<cell-key>/` or by patching the spec), then re-run. If the fix was in the worktree (e.g. the subagent left partial commits), you may want `--only <cell-key>` to avoid redundant PR-existence checks across many cells.
 
-- `--only platform-proto` — proto change has to ship before anything else can adopt the new bindings. Run the proto cell alone first, review the PR, merge it, then run the rest.
-- `--only java-sdk` — re-launch a single failed cell after fixing whatever broke. The dependency check still runs; if `java-sdk`'s `depends_on` failed earlier, the orchestrator will refuse rather than racing.
+**Staging a cell before its dependents** — `--only platform-proto` runs only the proto cell. Once its PR is reviewed and merged, re-run without `--only` and the orchestrator picks up from the next wave (proto cell already has a PR, so it's skipped; service/SDK cells proceed).
+
+**Forcing a re-run** — `--force` makes the orchestrator ignore existing PRs and dispatch a fresh subagent for every non-skipped cell. Combine with `--only` to force just one cell:
+```bash
+uv run otdf-sdk-mgr orchestrate run xtest/features/<name>.yaml --only java-sdk --force
+```
 
 ## Notes
 

otdf-sdk-mgr/src/otdf_sdk_mgr/cli_orchestrate.py

@@ -299,6 +299,19 @@ def build_prompt(spec: FeatureSpec, cell: Cell) -> str:
 PR_URL_RE = re.compile(r"https://github\.com/[^\s]+/pull/\d+")
 
 
+def check_existing_pr(repo: Path, branch: str) -> str | None:
+    """Return the URL of an open PR for this branch in the given repo, or None."""
+    result = subprocess.run(
+        ["gh", "pr", "list", "--head", branch, "--json", "url", "--jq", ".[0].url"],
+        cwd=str(repo),
+        capture_output=True,
+        text=True,
+    )
+    if result.returncode == 0 and result.stdout.strip():
+        return result.stdout.strip()
+    return None
+
+
 @dataclass
 class CellResult:
     cell: Cell
@@ -316,7 +329,16 @@ def run_cell(
     transcripts_dir: Path,
     timeout_s: int,
     model: str,
+    force: bool = False,
 ) -> CellResult:
+    # Idempotency: skip cells whose branch already has an open PR, unless --force.
+    if not force:
+        repo = OPENTDF_ROOT / (cell.path or "")
+        if repo.is_dir():
+            existing_pr = check_existing_pr(repo, cell.branch)
+            if existing_pr:
+                return CellResult(cell, Path(), Path(), True, existing_pr, None)
+
     try:
         wt = ensure_worktree(spec, cell)
     except Exception as e:
@@ -442,6 +464,10 @@ def run(
     model: Annotated[
         str, typer.Option("--model", help="Sub-agent model alias.")
     ] = "sonnet",
+    force: Annotated[
+        bool,
+        typer.Option("--force", help="Re-run cells even if they already have open PRs."),
+    ] = False,
     transcripts_dir: Annotated[
         Path,
         typer.Option(
@@ -481,14 +507,20 @@ def run(
             typer.echo(f"  Wave {i}:")
             for cell in wave:
                 if cell.key == "tests":
+                    existing = check_existing_pr(TESTS_REPO, cell.branch)
+                    pr_note = f" [PR EXISTS: {existing}]" if existing else ""
                     typer.echo(
                         f"    - {cell.key}: path=(tests repo) branch={cell.branch}"
-                        f" action=push+draft-PR"
+                        f" action=push+draft-PR{pr_note}"
                     )
                 else:
+                    repo = OPENTDF_ROOT / (cell.path or "")
+                    existing = check_existing_pr(repo, cell.branch) if repo.is_dir() else None
+                    pr_note = f" [PR EXISTS: {existing}]" if existing else ""
                     wt = worktree_for(spec, cell)
                     typer.echo(
-                        f"    - {cell.key}: path={cell.path} branch={cell.branch} worktree={wt}"
+                        f"    - {cell.key}: path={cell.path} branch={cell.branch}"
+                        f" worktree={wt}{pr_note}"
                     )
         return
 
@@ -512,6 +544,7 @@ def _dispatch(c: Cell) -> CellResult:
                 transcripts_dir=transcripts_dir,
                 timeout_s=timeout_s,
                 model=model,
+                force=force,
             )
 
         with concurrent.futures.ThreadPoolExecutor(max_workers=max(1, len(runnable))) as ex: