Implement LangGraph suspend-and-resume orchestration (2.4.2)

[leynos]

Summary

This branch implements roadmap task (2.4.2) by adding durable suspend-and-resume orchestration for structured generation runs. It persists workflow checkpoints before the side-effecting execution step, reuses the first checkpoint for repeated idempotency keys, and resumes from a checkpoint through TaskResumePort with an externally supplied action result.

Roadmap task: (2.4.2)
Execplan: docs/execplans/2-4-2-add-lang-graph-suspend-and-resume-orchestration.md

Review walkthrough

• Start with episodic/orchestration/langgraph.py for the checkpointing graph path, suspend result, and resume aggregation.
• Review episodic/orchestration/_dto.py and episodic/orchestration/_protocols.py for the provider-neutral checkpoint DTOs, ports, and idempotency key contract.
• Review episodic/canonical/storage/workflow_checkpoints.py and alembic/versions/20260508_000008_add_workflow_checkpoints.py for durable persistence.
• Finish with tests/test_generation_orchestration_langgraph.py, tests/canonical_storage/test_workflow_checkpoints.py, and tests/features/generation_orchestration.feature for unit, storage, property, and Vidai Mock behavioural coverage.

Validation

• make check-fmt: passed
• make typecheck: passed
• make lint: passed
• make test: passed, 446 passed, 3 skipped
• make markdownlint: passed, 0 error(s)
• make nixie: passed

Notes

make fmt was also run after documentation edits. It left Python files unchanged but still reports the repository-wide legacy Markdown MD013 formatter issue in unrelated documents. The explicit required gates above pass on the final branch state.

Summary by Sourcery

Add durable suspend-and-resume support to the LangGraph-based generation orchestration, including checkpoint persistence, idempotent workflow step keys, and resume handling, with tests, storage adapters, migration, and documentation updates.

New Features:

• Introduce resumable generation workflows that suspend before the first execution step, persist a checkpoint, and later resume to produce a final orchestration result.
• Expose orchestration DTOs and ports for workflow checkpoints, suspended workflow results, resume commands, and workflow step identities.
• Provide an in-memory checkpoint store and a SQLAlchemy-backed checkpoint store for orchestration checkpoints.

Enhancements:

• Extend the LangGraph generation graph to optionally use checkpointing, returning a suspended workflow result instead of executing the tool when a checkpoint port is configured.
• Add deterministic idempotency key construction for workflow steps and integrate it into checkpoint persistence and reuse logic.
• Export new orchestration types and helpers through the public orchestration module for broader reuse.
• Update developer, user, design, roadmap, and ADR documentation to describe durable generation checkpoints, idempotency behaviour, and testing locations.

Build:

• Add an Alembic migration and SQLAlchemy model for the workflow_checkpoints table used to store orchestration checkpoints.

Deployment:

• Wire the workflow checkpoint store into the SQLAlchemy unit of work and canonical storage exports so it is available to application composition roots.

Tests:

• Add unit tests for LangGraph suspension, checkpoint reuse, and resume aggregation using in-memory checkpoints and fake ports.
• Add SQLAlchemy-backed tests to verify workflow checkpoints persist across units of work and enforce idempotency on repeated saves.
• Introduce a property-based test to ensure workflow step idempotency keys are deterministic for identical inputs.
• Extend BDD scenarios with a Vidai Mock-based suspend-and-resume generation orchestration flow that asserts checkpoint reuse and model call ordering.

[sourcery-ai]

Sorry @leynos, you have reached your weekly rate limit of 2500000 diff characters.

Please try again later or upgrade to continue using Sourcery

[coderabbitai]

Warning

Rate limit exceeded

@lodyai[bot] has exceeded the limit for the number of commits that can be reviewed per hour. Please wait 38 seconds before requesting another review.

You’ve run out of usage credits. Purchase more in the billing tab.

⌛ How to resolve this issue?

After the wait time has elapsed, a review can be triggered using the @coderabbitai review command as a PR comment. Alternatively, push new commits to this PR.

We recommend that you space out your commits to avoid hitting the rate limit.

🚦 How do rate limits work?

CodeRabbit enforces hourly rate limits for each developer per organization.

Our paid plans have higher rate limits than the trial, open-source and free plans. In all cases, we re-allow further reviews after a brief timeout.

Please see our FAQ for further information.

ℹ️ Review info

⚙️ Run configuration

Configuration used: Organization UI

Review profile: ASSERTIVE

Plan: Pro Plus

Run ID: 2886a8f6-2b5b-4d30-b2dc-99850aa7eb72

📥 Commits

Reviewing files that changed from the base of the PR and between 20c0abc and d5989e0.

📒 Files selected for processing (2)

• docs/adr/adr-007-durable-generation-checkpoints.md
• episodic/orchestration/langgraph.py

Walkthrough

Persist planner state before side effects, return a SuspendedWorkflowResult, provide in-memory and SQLAlchemy checkpoint adapters with idempotent save_or_reuse/mark_resumed semantics, wire suspend-before-execute and resume entrypoints into LangGraph, add Alembic migration and unit-of-work wiring, and cover behaviour with unit, property, snapshot, BDD and SQLAlchemy tests plus ADR/ExecPlan and guide updates.

Changes

Durable Suspend-and-Resume Orchestration

Layer / File(s)	Summary
Orchestration DTOs and Protocols episodic/orchestration/_dto.py, episodic/orchestration/_protocols.py	Define WorkflowCheckpoint, SuspendedWorkflowResult, ResumeWorkflowCommand, WorkflowStepIdentity, build_workflow_step_idempotency_key, and CheckpointPort/TaskResumePort protocols. Make PlannerResult.usage and GenerationOrchestrationResult.planner_usage optional.
In-Memory Checkpoint Store episodic/orchestration/checkpoints.py	Implement InMemoryCheckpointStore with asyncio.Lock-protected first-write-wins idempotency, injected time_provider for deterministic test timestamps, and concurrent deduplication support.
SQLAlchemy ORM Model and Store episodic/canonical/storage/models.py, episodic/canonical/storage/workflow_checkpoints.py	Define WORKFLOW_CHECKPOINT_STATUS and WorkflowCheckpointRecord ORM; implement SqlAlchemyWorkflowCheckpointStore mapping ORM ↔ DTO with nested-transaction insert-then-load idempotency and mark_resumed.
Database Migration and UnitOfWork Wiring alembic/versions/20260508_000008_add_workflow_checkpoints.py, episodic/canonical/storage/uow.py	Add Alembic migration creating workflow_checkpoints table, enum type, index and update trigger; wire SqlAlchemyWorkflowCheckpointStore into SqlAlchemyUnitOfWork.__aenter__.
LangGraph Suspend-and-Resume Integration episodic/orchestration/langgraph.py	Add checkpoint payload serialisers/deserialisers with runtime validation; extend GenerationGraphState with suspended_result; implement suspend-before-execute to persist/reuse checkpoints and return SuspendedWorkflowResult; add resume_generation_orchestration to load checkpoint, validate planner payload, call TaskResumePort.resume, aggregate action result and mark checkpoint as resumed; toggle graph wiring with checkpoint_port.
Module Exports and Public API episodic/orchestration/__init__.py, episodic/orchestration/generation.py, episodic/canonical/storage/__init__.py	Re-export InMemoryCheckpointStore, DTOs, ports, build_workflow_step_idempotency_key, resume_generation_orchestration, SqlAlchemyWorkflowCheckpointStore and WorkflowCheckpointRecord.
LangGraph Unit Tests (suspend/resume) tests/test_generation_orchestration_langgraph.py	Verify suspend before tool execution with matching idempotency key, checkpoint reuse on repeated invocations, resume_generation_orchestration aggregates token usage and marks resumed, rejects multi-step checkpoints, raises on unknown/invalid checkpoints, and verifies in-memory concurrent deduplication and mark_resumed error.
SQLAlchemy Checkpoint Tests tests/canonical_storage/test_workflow_checkpoints.py	Verify checkpoint persistence across unit-of-work boundaries, idempotency key reuse, null returns for missing lookups, mark_resumed persistence and unknown-id errors.
Property-Based and Snapshot Tests tests/test_orchestration_properties.py, tests/test_generation_orchestration_snapshots.py	Verify idempotency-key determinism and negative-attempt rejection; verify payload round-trips for plans, planner results and action results; add Syrupy snapshot for checkpoint payload; update planner test for optional usage.
BDD Acceptance tests/features/generation_orchestration.feature, tests/steps/test_generation_orchestration_steps.py	Add Gherkin scenario for suspend/resume flow. Implement step exercising graph twice, verifying checkpoint persistence, executing action, resuming and aggregating result. Verify checkpoint reuse and model invocation order.
ADR, ExecPlan, Guides and Roadmap docs/adr/adr-007-durable-generation-checkpoints.md, docs/execplans/2-4-2-add-lang-graph-suspend-and-resume-orchestration.md, docs/developers-guide.md, docs/episodic-podcast-generation-system-design.md, docs/users-guide.md, docs/roadmap.md	Formalise checkpoint approach in ADR-007 with idempotency semantics, adapter concurrency behaviour and consequences. Complete ExecPlan 2.4.2 with goals, constraints, risks, progress, surprises, decision log and revision notes. Update guides with checkpoint modules, maintainer rules, system design section and users-facing resumable orchestration docs. Mark roadmap item 2.4.2 complete.

Possibly related issues

• Document and test atomicity boundaries for checkpoint suspend/resume operations #91: Addresses duplicate-key/atomicity concerns in SqlAlchemyWorkflowCheckpointStore.save_or_reuse and mark_resumed by using nested transactions and re-query-on-IntegrityError.

Poem

Persist the plan, then pause the run,
Save the key so work is not redone,
Resume with result, aggregate the art,
Mark it resumed and play your part,
Checkpoint and retry — orchestration done.

---

Caution

Pre-merge checks failed

Please resolve all errors before merging. Addressing warnings is optional.

• Ignore

❌ Failed checks (2 errors, 6 warnings)

Check name	Status	Explanation	Resolution
Testing (Overall)	❌ Error	Tests cover core behaviour well but critical gaps remain: invalid enum values in payloads not tested; enum parsing raises ValueError not TypeError; property tests never generate invalid enums.	Add tests with invalid action_kind/model_tier strings in payloads, verifying TypeError. Expand Hypothesis to generate invalid enum values. Implement _required_enum helper to normalise ValueError→TypeError.
Unit Architecture	❌ Error	Enum parsing raises ValueError instead of documented TypeError. Missing _required_enum helper. Test coverage gap for invalid enum values. ADR acronym unexpanded.	Add _required_enum helper normalising ValueError to TypeError. Update all enum instantiations in payload deserialization. Add test for invalid enum payloads. Expand ADR title acronym.
Developer Documentation	⚠️ Warning	ADR-007 title does not expand the acronym on first use. Review feedback required expansion to match ADR-003 precedent: "Architectural Decision Record (ADR-007)".	Update docs/adr/adr-007-durable-generation-checkpoints.md heading from # ADR-007: Durable generation checkpoints to # Architectural Decision Record (ADR-007): Durable generation checkpoints per coding guidelines.
Testing (Unit And Behavioural)	⚠️ Warning	Tests lack coverage for invalid enum values in checkpoint payloads. ActionKind/ModelTier construction raises ValueError instead of documented TypeError.	Add test with invalid enum values in checkpoint payloads (e.g., action_kind="INVALID") verifying TypeError. Alternatively implement _required_enum helper to normalise ValueError to TypeError.
Testing (Property / Proof)	⚠️ Warning	PR introduces seven major checkpoint invariants. Only three have property tests. Missing: status transitions, concurrent orderings, checkpoint immutability, and mark_resumed error paths.	Add Hypothesis property tests for status validity, multi-ordering concurrency (asyncio+Hypothesis), checkpoint persistence round-trips, and mark_resumed error cases.
Domain Architecture	⚠️ Warning	Enum parsing raises ValueError instead of promised TypeError. Lines 160-161, 251-252 ActionKind/ModelTier constructors can fail with ValueError, violating documented resume error contract.	Wrap enum construction with _required_enum() helper to convert ValueError to TypeError and honour the documented exception contract.
Observability	⚠️ Warning	Missing error logging in resume and mark_resumed error paths. No metrics for checkpoint operations or distributed tracing.	Add _log_event calls for multi-step validation, planner deserialisation, and resume_port failures. Add error logging to both mark_resumed implementations. Add metrics for checkpoint persistence and idempotency conflicts.
Concurrency And State	⚠️ Warning	InMemoryCheckpointStore.get() lacks lock acquisition, allowing stale reads during concurrent writes. No concurrent SQL tests across UoW boundaries. Concurrency model undocumented.	Acquire lock in InMemory get methods; add concurrent SQL tests across UoW boundaries; document concurrency semantics in CheckpointPort protocol; add property tests for concurrent operation interleaving.

✅ Passed checks (10 passed)

Check name	Status	Explanation
Title check	✅ Passed	The title fully aligns with the PR's primary change: implementing LangGraph suspend-and-resume orchestration with the required roadmap reference (2.4.2).
Description check	✅ Passed	The description comprehensively documents the changeset's scope, objectives, and implementation across all major areas: orchestration, DTOs, protocols, storage, idempotency, public APIs, tests, and documentation.
Docstring Coverage	✅ Passed	Docstring coverage is 91.49% which is sufficient. The required threshold is 80.00%.
Linked Issues check	✅ Passed	Check skipped because no linked issues were found for this pull request.
Out of Scope Changes check	✅ Passed	Check skipped because no linked issues were found for this pull request.
User-Facing Documentation	✅ Passed	Documentation complete. Resumable orchestration section properly added to docs/users-guide.md and README. Appropriately pitched for operators. No breaking changes. No language sync needed.
Module-Level Documentation	✅ Passed	All modules carry substantive docstrings explaining purpose, utility, function, and relationships to other components.
Testing (Compile-Time / Ui)	✅ Passed	Snapshot tests appropriately capture checkpoint serialisation with meaningful expectations. Property-based round-trip tests verify payload correctness. Migrations are hand-written, not generated code.
Security And Privacy	✅ Passed	PR introduces no security or privacy risks. Database queries are parameterised, inputs validated, errors handle safely, logging is controlled, and no sensitive data is exposed.
Performance And Resource Use	✅ Passed	No regressions detected. Payload cloning is intentional for immutability. In-memory store is test-only. Database operations are appropriately batched and occur once per suspend/resume cycle.

✨ Finishing Touches

🧪 Generate unit tests (beta)

• Create PR with unit tests
• Commit unit tests in branch 2-4-2-add-lang-graph-suspend-and-resume-orchestration

---

_{Comment @coderabbitai help to get the list of available commands and usage tips.}

[leynos]

@coderabbitai Please suggest a fix for this issue and supply a prompt for an AI coding agent to enable it to apply the fix. Include the file and symbol names indicated in the issue at the head of your response.

tests/test_orchestration_properties.py

Comment on lines +121 to +144

def test_step_idempotency_keys_are_deterministic(
    workflow_id: str,
    workflow_type: str,
    step_name: str,
    action_id: str,
    attempt: int,
) -> None:
    """Property test: identical workflow step inputs produce identical keys."""
    first = build_workflow_step_idempotency_key(
        workflow_id=workflow_id,
        workflow_type=workflow_type,
        step_name=step_name,
        action_id=action_id,
        attempt=attempt,
    )
    second = build_workflow_step_idempotency_key(
        workflow_id=workflow_id,
        workflow_type=workflow_type,
        step_name=step_name,
        action_id=action_id,
        attempt=attempt,
    )
    assert second == first
    assert first.endswith(f":{attempt}")

❌ New issue: Excess Number of Function Arguments
test_step_idempotency_keys_are_deterministic has 5 arguments, max arguments = 4

[leynos]

@coderabbitai Please suggest a fix for this issue and supply a prompt for an AI coding agent to enable it to apply the fix. Include the file and symbol names indicated in the issue at the head of your response.

episodic/orchestration/_dto.py

Comment on lines +474 to +493

def build_workflow_step_idempotency_key(  # noqa: PLR0913
    *,
    workflow_id: str,
    workflow_type: str,
    step_name: str,
    action_id: str,
    attempt: int = 0,
) -> str:
    """Build the deterministic idempotency key for a suspendable workflow step."""
    if attempt < 0:
        msg = "attempt must be greater than or equal to zero."
        raise ValueError(msg)
    parts = (
        _normalize_non_empty_text(workflow_id, "workflow_id"),
        _normalize_non_empty_text(workflow_type, "workflow_type"),
        _normalize_non_empty_text(step_name, "step_name"),
        _normalize_non_empty_text(action_id, "action_id"),
        str(attempt),
    )
    return ":".join(parts)

❌ New issue: Excess Number of Function Arguments
build_workflow_step_idempotency_key has 5 arguments, max arguments = 4

[sourcery-ai]

Reviewer's Guide

Implements durable suspend-and-resume orchestration for structured generation by introducing checkpoint DTOs and ports, a LangGraph path that suspends before execution and resumes from persisted checkpoints, plus in-memory and SQLAlchemy-backed checkpoint stores, tests, and documentation/roadmap updates.

Sequence diagram for suspend-before-execute generation orchestration

sequenceDiagram
    actor Client
    participant Orchestrator as StructuredPlanningOrchestrator
    participant LangGraph as GenerationGraph
    participant Planner as PlannerPort
    participant Checkpoints as CheckpointPort

    Client->>Orchestrator: start_generation(request)
    Orchestrator->>LangGraph: run(request, checkpoint_port)

    LangGraph->>Planner: plan(request)
    Planner-->>LangGraph: PlannerResult

    LangGraph->>Checkpoints: get_by_idempotency_key(step_idempotency_key)
    Checkpoints-->>LangGraph: WorkflowCheckpoint | None

    alt checkpoint does not exist
        LangGraph->>Checkpoints: save(WorkflowCheckpoint)
        Checkpoints-->>LangGraph: WorkflowCheckpoint
    end

    LangGraph-->>Orchestrator: SuspendedWorkflowResult
    Orchestrator-->>Client: SuspendedWorkflowResult

Loading

Sequence diagram for resuming a suspended generation workflow

sequenceDiagram
    actor Client
    participant Orchestrator as GenerationOrchestrationLayer
    participant Checkpoints as CheckpointPort
    participant ResumePort as TaskResumePort

    Client->>Orchestrator: resume_generation(ResumeWorkflowCommand)

    Orchestrator->>Checkpoints: get(checkpoint_id)
    Checkpoints-->>Orchestrator: WorkflowCheckpoint | None

    alt checkpoint unknown
        Orchestrator-->>Client: ValueError
    else checkpoint found
        Orchestrator->>ResumePort: resume(ResumeWorkflowCommand)
        ResumePort-->>Orchestrator: ActionExecutionResult

        Orchestrator->>Orchestrator: build_generation_result(PlannerResult, ActionExecutionResult)
        Orchestrator-->>Client: GenerationOrchestrationResult
    end

Loading

ER diagram for workflow_checkpoints table

erDiagram
    WORKFLOW_CHECKPOINTS {
        uuid id PK
        string workflow_id
        string workflow_type
        string step_name
        string idempotency_key
        jsonb payload
        string status
        datetime created_at
        datetime updated_at
    }

Loading

Class diagram for orchestration checkpoint DTOs, ports, and adapters

classDiagram
    class WorkflowCheckpoint {
        +str checkpoint_id
        +str workflow_id
        +str workflow_type
        +str step_name
        +str idempotency_key
        +dict~str, object~ payload
        +str status
        +datetime created_at
        +datetime updated_at
        +__post_init__() void
    }

    class SuspendedWorkflowResult {
        +str checkpoint_id
        +str workflow_id
        +str step_name
        +str idempotency_key
        +__post_init__() void
    }

    class ResumeWorkflowCommand {
        +str checkpoint_id
        +ActionExecutionResult result
        +__post_init__() void
    }

    class WorkflowStepIdentity {
        +str workflow_id
        +str workflow_type
        +str step_name
        +str action_id
        +__post_init__() void
    }

    class GenerationGraphState {
        +GenerationOrchestrationRequest request
        +PlannerResult planner_result
        +tuple~ActionExecutionResult~ action_results
        +GenerationOrchestrationResult orchestration_result
        +SuspendedWorkflowResult suspended_result
    }

    class CheckpointPort {
        <<protocol>>
        +get(checkpoint_id str) WorkflowCheckpoint
        +get_by_idempotency_key(idempotency_key str) WorkflowCheckpoint
        +save(checkpoint WorkflowCheckpoint) WorkflowCheckpoint
    }

    class TaskResumePort {
        <<protocol>>
        +resume(command ResumeWorkflowCommand) ActionExecutionResult
    }

    class InMemoryCheckpointStore {
        +dict~str, WorkflowCheckpoint~ _by_id
        +dict~str, str~ _by_key
        +get(checkpoint_id str) WorkflowCheckpoint
        +get_by_idempotency_key(idempotency_key str) WorkflowCheckpoint
        +save(checkpoint WorkflowCheckpoint) WorkflowCheckpoint
    }

    class SqlAlchemyWorkflowCheckpointStore {
        -AsyncSession _session
        +get(checkpoint_id str) WorkflowCheckpoint
        +get_by_idempotency_key(idempotency_key str) WorkflowCheckpoint
        +save(checkpoint WorkflowCheckpoint) WorkflowCheckpoint
    }

    class WorkflowCheckpointRecord {
        +uuid id
        +str workflow_id
        +str workflow_type
        +str step_name
        +str idempotency_key
        +dict~str, object~ payload
        +str status
        +datetime created_at
        +datetime updated_at
    }

    class Functions {
        +build_workflow_step_idempotency_key(step WorkflowStepIdentity, attempt int) str
        +resume_generation_orchestration(checkpoint_port CheckpointPort, resume_port TaskResumePort, command ResumeWorkflowCommand) GenerationOrchestrationResult
    }

    CheckpointPort <|.. InMemoryCheckpointStore
    CheckpointPort <|.. SqlAlchemyWorkflowCheckpointStore

    InMemoryCheckpointStore o--> WorkflowCheckpoint
    SqlAlchemyWorkflowCheckpointStore o--> WorkflowCheckpointRecord

    GenerationGraphState o--> SuspendedWorkflowResult
    GenerationGraphState o--> PlannerResult
    GenerationGraphState o--> ActionExecutionResult
    GenerationGraphState o--> GenerationOrchestrationResult

    Functions ..> WorkflowStepIdentity
    Functions ..> CheckpointPort
    Functions ..> TaskResumePort
    WorkflowCheckpointRecord --> WorkflowCheckpoint

Loading

File-Level Changes

Change	Details	Files
Add orchestration DTOs and idempotency helpers for workflow checkpoints and resume commands.	Introduce WorkflowCheckpoint, SuspendedWorkflowResult, ResumeWorkflowCommand, and WorkflowStepIdentity dataclasses with validation and normalization Add build_workflow_step_idempotency_key helper and corresponding Hypothesis property test for deterministic keys Re-export new orchestration DTOs and helpers through public orchestration modules	episodic/orchestration/_dto.py episodic/orchestration/generation.py episodic/orchestration/__init__.py tests/test_orchestration_properties.py
Extend LangGraph orchestration to support suspend-before-execute and resume-from-checkpoint flows.	Add serialization helpers to convert planner and action results to and from JSON-compatible payloads for checkpoint storage Implement _suspend_execute_node that persists a checkpoint before the first action execution using a CheckpointPort and returns SuspendedWorkflowResult Add resume_generation_orchestration function that loads a checkpoint, resumes the external task via TaskResumePort, and builds the final GenerationOrchestrationResult Update build_generation_orchestration_graph to optionally wire a checkpointing execute node that ends the graph when a CheckpointPort is provided, while preserving existing behaviour when it is not Add unit tests to verify suspension, checkpoint reuse for same step, and resume aggregation semantics	episodic/orchestration/langgraph.py tests/test_generation_orchestration_langgraph.py
Define provider-neutral ports and in-memory adapter for checkpoint persistence and task resumption.	Add CheckpointPort and TaskResumePort protocols with get, get_by_idempotency_key, save, and resume contracts Implement InMemoryCheckpointStore as a simple async checkpoint adapter with idempotent save semantics for tests Use new ports and adapter in BDD steps and tests to drive suspend-and-resume flows	episodic/orchestration/_protocols.py episodic/orchestration/checkpoints.py tests/steps/test_generation_orchestration_steps.py
Introduce durable SQLAlchemy-backed checkpoint storage and wire it into canonical storage.	Add WorkflowCheckpointRecord SQLAlchemy model and Alembic migration to create workflow_checkpoints table with unique idempotency_key and workflow_id index Implement SqlAlchemyWorkflowCheckpointStore that maps between WorkflowCheckpointRecord and WorkflowCheckpoint, providing get, get_by_idempotency_key, and idempotent save Expose SqlAlchemyWorkflowCheckpointStore and WorkflowCheckpointRecord via canonical storage package and UnitOfWork Add py-pglite-backed tests to verify persistence across unit-of-work instances and idempotency-key reuse semantics	episodic/canonical/storage/models.py episodic/canonical/storage/workflow_checkpoints.py episodic/canonical/storage/__init__.py episodic/canonical/storage/uow.py tests/canonical_storage/test_workflow_checkpoints.py alembic/versions/20260508_000008_add_workflow_checkpoints.py
Enhance behavioural tests and documentation to cover suspend-and-resume orchestration and mark roadmap item complete.	Extend BDD feature and step definitions to cover LangGraph suspend-before-execute, resume via ResumeWorkflowCommand, and checkpoint reuse assertions using Vidai Mock Adjust orchestration request assertions to account for extra planning call in suspend/resume scenario Document checkpointing architecture, ports, adapter locations, and idempotency-key usage in developer and system design guides, and describe resumable orchestration in the user guide Add ADR-006 describing durable generation checkpoints and mark roadmap item 2.4.2 as completed	tests/steps/test_generation_orchestration_steps.py tests/features/generation_orchestration.feature docs/developers-guide.md docs/episodic-podcast-generation-system-design.md docs/users-guide.md docs/roadmap.md docs/adr/adr-006-durable-generation-checkpoints.md docs/execplans/2-4-2-add-lang-graph-suspend-and-resume-orchestration.md

---

Tips and commands

Interacting with Sourcery

• Trigger a new review: Comment @sourcery-ai review on the pull request.
• Continue discussions: Reply directly to Sourcery's review comments.
• Generate a GitHub issue from a review comment: Ask Sourcery to create an
  issue from a review comment by replying to it. You can also reply to a
  review comment with @sourcery-ai issue to create an issue from it.
• Generate a pull request title: Write @sourcery-ai anywhere in the pull
  request title to generate a title at any time. You can also comment
  @sourcery-ai title on the pull request to (re-)generate the title at any time.
• Generate a pull request summary: Write @sourcery-ai summary anywhere in
  the pull request body to generate a PR summary at any time exactly where you
  want it. You can also comment @sourcery-ai summary on the pull request to
  (re-)generate the summary at any time.
• Generate reviewer's guide: Comment @sourcery-ai guide on the pull
  request to (re-)generate the reviewer's guide at any time.
• Resolve all Sourcery comments: Comment @sourcery-ai resolve on the
  pull request to resolve all Sourcery comments. Useful if you've already
  addressed all the comments and don't want to see them anymore.
• Dismiss all Sourcery reviews: Comment @sourcery-ai dismiss on the pull
  request to dismiss all existing Sourcery reviews. Especially useful if you
  want to start fresh with a new review - don't forget to comment
  @sourcery-ai review to trigger a new review!

Customizing Your Experience

Access your dashboard to:

• Enable or disable review features such as the Sourcery-generated pull request
  summary, the reviewer's guide, and others.
• Change the review language.
• Add, remove or edit custom review instructions.
• Adjust other review settings.

Getting Help

• Contact our support team for questions or feedback.
• Visit our documentation for detailed guides and information.
• Keep in touch with the Sourcery team by following us on X/Twitter, LinkedIn or GitHub.

[chatgpt-codex-connector]

💡 Codex Review

Here are some automated review suggestions for this pull request.

Reviewed commit: ce9b04d236

ℹ️ About Codex in GitHub

Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you

• Open a pull request for review
• Mark a draft as ready
• Comment "@codex review".

If Codex has suggestions, it will comment; otherwise it will react with 👍.

Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".

[coderabbitai]

Actionable comments posted: 4

🤖 Prompt for all review comments with AI agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

Inline comments:
In `@docs/developers-guide.md`:
- Around line 650-651: Update the documentation for
build_workflow_step_idempotency_key to state that it accepts a
WorkflowStepIdentity and an attempt count rather than five scalar arguments;
specifically, reword the sentence to explain that callers should construct a
WorkflowStepIdentity (containing workflow id, workflow type, step name, and
action id) and pass a separate attempt parameter for the retry count when
invoking build_workflow_step_idempotency_key.

In `@docs/execplans/2-4-2-add-lang-graph-suspend-and-resume-orchestration.md`:
- Around line 525-530: The focused pytest command references the wrong steps
module name; change the test file in the command from
tests/steps/test_generation_suspend_resume_steps.py to
tests/steps/test_generation_orchestration_steps.py so the documented run loads
the implemented suspend/resume BDD steps; update the command string shown (the
PYO3... uv run pytest ... line) to use the correct test module name.

In `@episodic/orchestration/_dto.py`:
- Line 3: Remove the redundant future-import at the top of the module: delete
the line "from __future__ import annotations" from the top of episodic
orchestration DTO module (the module-level import in _dto.py) so the file no
longer contains that forbidden future import; no other changes are required.

In `@episodic/orchestration/langgraph.py`:
- Around line 346-360: The resume_generation_orchestration function currently
assumes resume_port.resume(command) returns a single ActionExecutionResult and
builds the final result with build_generation_result(planner_result,
(action_result,)); update the function docstring to explicitly state this
single-action-per-suspend assumption (and that future changes to allow multiple
actions must update this code path), referencing
resume_generation_orchestration, resume_port.resume, planner_result, and
build_generation_result so maintainers can find and adjust the logic if the
model changes.

🪄 Autofix (Beta)

Fix all unresolved CodeRabbit comments on this PR:

• Push a commit to this branch (recommended)
• Create a new PR with the fixes

---

ℹ️ Review info

⚙️ Run configuration

Configuration used: Organization UI

Review profile: ASSERTIVE

Plan: Pro Plus

Run ID: 503c444e-13bd-4525-8551-2aa4d810828e

📥 Commits

Reviewing files that changed from the base of the PR and between 4bf5e9e and ce9b04d.

📒 Files selected for processing (22)

• alembic/versions/20260508_000008_add_workflow_checkpoints.py
• docs/adr/adr-006-durable-generation-checkpoints.md
• docs/developers-guide.md
• docs/episodic-podcast-generation-system-design.md
• docs/execplans/2-4-2-add-lang-graph-suspend-and-resume-orchestration.md
• docs/roadmap.md
• docs/users-guide.md
• episodic/canonical/storage/__init__.py
• episodic/canonical/storage/models.py
• episodic/canonical/storage/uow.py
• episodic/canonical/storage/workflow_checkpoints.py
• episodic/orchestration/__init__.py
• episodic/orchestration/_dto.py
• episodic/orchestration/_protocols.py
• episodic/orchestration/checkpoints.py
• episodic/orchestration/generation.py
• episodic/orchestration/langgraph.py
• tests/canonical_storage/test_workflow_checkpoints.py
• tests/features/generation_orchestration.feature
• tests/steps/test_generation_orchestration_steps.py
• tests/test_generation_orchestration_langgraph.py
• tests/test_orchestration_properties.py

[coderabbitai]

Actionable comments posted: 6

🤖 Prompt for all review comments with AI agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

Inline comments:
In `@docs/adr/adr-006-durable-generation-checkpoints.md`:
- Line 37: Update the British spelling "serialises" to the Oxford -ize form
"serializes" in the sentence that reads "The in-memory adapter serialises
`save(...)` mutations with an `asyncio.Lock`" so the documentation matches the
project's en-GB-oxendict convention; search for the phrase containing
`save(...)` and `asyncio.Lock` and replace "serialises" with "serializes".

In `@docs/developers-guide.md`:
- Around line 625-626: Replace the phrase "independent from" with the correct
collocation "independent of" in the sentence containing "checkpoint, and resume
ports that keep graph policy independent from storage, queue, and provider
adapters." Locate that exact phrase ("independent from") in the
docs/developers-guide.md content and update it to "independent of" so the
sentence reads "...keep graph policy independent of storage, queue, and provider
adapters."

In `@docs/episodic-podcast-generation-system-design.md`:
- Around line 471-525: Add short screen-reader descriptions immediately before
each Mermaid sequenceDiagram block (the "Suspend flow" diagram that begins with
"Client->>Orchestrator: start_generation(request)" and the "Resume flow" diagram
that begins with "Client->>Orchestrator:
resume_generation(ResumeWorkflowCommand)"); insert a one- or two-sentence
plain-text line labeled like "Screen reader description:" summarizing the
diagram intent (e.g., checkpoint persistence before side-effecting execution for
the first, and checkpoint lookup/reject-or-resume for the second) so assistive
readers get context before the mermaid code blocks.

In `@docs/execplans/2-4-2-add-lang-graph-suspend-and-resume-orchestration.md`:
- Line 681: The document uses the non-Oxford spelling "serialised" in the phrase
"serialised in-memory"; update every occurrence of "serialised" to the Oxford
spelling "serialized" (e.g., change "serialised in-memory checkpoint clock" to
"serialized in-memory checkpoint clock") to match the en-GB-oxendict `-ize`
convention; search for the token "serialised" in this file (and nearby lines)
and replace with "serialized" and run a quick spellcheck to ensure consistency.

In `@episodic/orchestration/langgraph.py`:
- Around line 401-402: The code reads payload = checkpoint.payload and then uses
payload["planner_result"] which can raise KeyError for malformed checkpoints;
update the block that calls _planner_result_from_payload to guard for a missing
"planner_result" key: check for "planner_result" in payload (or use
payload.get("planner_result")) and if missing raise a TypeError with a clear
message (or convert the KeyError to TypeError) so the function's exception
behavior matches its docstring; reference the checkpoint.payload access and the
call to _planner_result_from_payload when making the change.
- Around line 297-370: The _suspend_execute_node function is too long; extract
the checkpoint construction and persistence (the block that creates
dto.WorkflowCheckpoint and calls checkpoint_port.save when existing is None)
into a new helper like _create_and_save_checkpoint that accepts
(checkpoint_port, request, planner_result, workflow_id, workflow_type, action,
idempotency_key) and returns the saved checkpoint object; inside the helper
build the payload (using _planner_result_to_payload and request fields),
generate checkpoint_id via uuid.uuid4(), call
checkpoint_port.save(dto.WorkflowCheckpoint(...)) and return the result; then
replace the original inline block in _suspend_execute_node with a call to this
helper and preserve the existing/reused_checkpoint logic and logging; reference
dto.WorkflowCheckpoint, dto.build_workflow_step_idempotency_key,
checkpoint_port.save, and _planner_result_to_payload when implementing.

🪄 Autofix (Beta)

Fix all unresolved CodeRabbit comments on this PR:

• Push a commit to this branch (recommended)
• Create a new PR with the fixes

---

ℹ️ Review info

⚙️ Run configuration

Configuration used: Organization UI

Review profile: ASSERTIVE

Plan: Pro Plus

Run ID: 27990eef-a6fd-42fa-8c21-11a84816b9c3

📥 Commits

Reviewing files that changed from the base of the PR and between ce9b04d and 6e2f7e7.

📒 Files selected for processing (13)

• docs/adr/adr-006-durable-generation-checkpoints.md
• docs/developers-guide.md
• docs/episodic-podcast-generation-system-design.md
• docs/execplans/2-4-2-add-lang-graph-suspend-and-resume-orchestration.md
• episodic/canonical/storage/workflow_checkpoints.py
• episodic/orchestration/_dto.py
• episodic/orchestration/_protocols.py
• episodic/orchestration/checkpoints.py
• episodic/orchestration/langgraph.py
• tests/__snapshots__/test_generation_orchestration_snapshots.ambr
• tests/test_generation_orchestration_langgraph.py
• tests/test_generation_orchestration_snapshots.py
• tests/test_orchestration_properties.py

[coderabbitai]

Actionable comments posted: 2

♻️ Duplicate comments (1)

episodic/orchestration/langgraph.py (1)

479-480: ⚠️ Potential issue | 🟡 Minor | ⚡ Quick win

Guard against missing planner_result key in checkpoint payload.

Line 480 accesses payload["planner_result"] directly. If the checkpoint payload is malformed, this raises KeyError, but the docstring documents only TypeError. Either catch KeyError and raise TypeError, or check for the key's presence first.

🛡️ Proposed fix

     payload = checkpoint.payload
+    if "planner_result" not in payload:
+        msg = "checkpoint payload missing required key: planner_result"
+        raise TypeError(msg)
     planner_result = _planner_result_from_payload(payload["planner_result"])

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@episodic/orchestration/langgraph.py` around lines 479 - 480, The code
accesses payload["planner_result"] without guarding for a missing key which can
raise KeyError; update the checkpoint handling in the function that sets payload
= checkpoint.payload to validate the key before use: either check
'planner_result' in payload and raise a TypeError with the documented message if
absent, or wrap the call to
_planner_result_from_payload(payload["planner_result"]) in a try/except catching
KeyError and re-raising a TypeError; ensure you reference the existing symbols
payload, planner_result, _planner_result_from_payload and checkpoint in your
change so behavior and docstring remain consistent.

🤖 Prompt for all review comments with AI agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

Inline comments:
In `@episodic/canonical/storage/workflow_checkpoints.py`:
- Line 11: Remove the unnecessary future import in the new module: delete the
line "from __future__ import annotations" from
episodic.canonical.storage.workflow_checkpoints (the module-level import at top
of file) so the file conforms to the Python ≥3.14 baseline and project coding
guidelines.

In `@tests/canonical_storage/test_workflow_checkpoints.py`:
- Line 3: Remove the unnecessary future import in the test module by deleting
the line "from __future__ import annotations" in
tests/canonical_storage/test_workflow_checkpoints.py; since the project targets
Python ≥3.14 and PEP 563 behavior is the default, simply remove that import
statement so the module no longer contains it.

---

Duplicate comments:
In `@episodic/orchestration/langgraph.py`:
- Around line 479-480: The code accesses payload["planner_result"] without
guarding for a missing key which can raise KeyError; update the checkpoint
handling in the function that sets payload = checkpoint.payload to validate the
key before use: either check 'planner_result' in payload and raise a TypeError
with the documented message if absent, or wrap the call to
_planner_result_from_payload(payload["planner_result"]) in a try/except catching
KeyError and re-raising a TypeError; ensure you reference the existing symbols
payload, planner_result, _planner_result_from_payload and checkpoint in your
change so behavior and docstring remain consistent.

🪄 Autofix (Beta)

Fix all unresolved CodeRabbit comments on this PR:

• Push a commit to this branch (recommended)
• Create a new PR with the fixes

---

ℹ️ Review info

⚙️ Run configuration

Configuration used: Organization UI

Review profile: ASSERTIVE

Plan: Pro Plus

Run ID: 3cd9fbe2-81ee-4aba-9d81-e0e7e4e99eba

📥 Commits

Reviewing files that changed from the base of the PR and between 6e2f7e7 and 6a80044.

📒 Files selected for processing (12)

• docs/adr/adr-006-durable-generation-checkpoints.md
• docs/episodic-podcast-generation-system-design.md
• docs/execplans/2-4-2-add-lang-graph-suspend-and-resume-orchestration.md
• episodic/canonical/storage/workflow_checkpoints.py
• episodic/orchestration/_dto.py
• episodic/orchestration/_protocols.py
• episodic/orchestration/checkpoints.py
• episodic/orchestration/langgraph.py
• tests/canonical_storage/test_workflow_checkpoints.py
• tests/test_generation_orchestration_langgraph.py
• tests/test_orchestration_planner.py
• tests/test_orchestration_properties.py

[coderabbitai]

Caution

Some comments are outside the diff and can’t be posted inline due to platform limitations.

⚠️ Outside diff range comments (1)

docs/execplans/2-4-2-add-lang-graph-suspend-and-resume-orchestration.md (1)

689-718: ⚠️ Potential issue | 🟡 Minor | ⚡ Quick win

Use Oxford spelling for -ize/-ization/-izing forms.

The en-GB-oxendict convention requires the -ize suffix. Update the following:

• Line 691: "serialised" → "serialized"
• Line 698: "deserialisation" → "deserialization"
• Line 712: "finalising" → "finalizing"

Proposed fix

 Revision note 2026-05-10: Follow-up review fixes added richer suspend/resume
 module documentation, documented `resume_generation_orchestration` error paths,
-injected the in-memory checkpoint clock, serialised in-memory checkpoint saves
+injected the in-memory checkpoint clock, serialized in-memory checkpoint saves
 with an `asyncio.Lock`, and changed the SQLAlchemy checkpoint adapter to insert
 first and query only after duplicate-key conflicts. Added missing
 unknown-checkpoint, concurrent in-memory save, checkpoint payload property, and
 checkpoint payload snapshot coverage.

 Revision note 2026-05-10: Code review follow-up made checkpoint planner payload
-deserialisation symmetric for optional planner usage, added the explicit
+deserialization symmetric for optional planner usage, added the explicit
 `CheckpointPort.mark_resumed()` status transition after successful resume,
 fixed the SQLAlchemy checkpoint store runtime annotation, and covered negative
 idempotency attempts plus SQL not-found checkpoint lookups.

...

 Revision note 2026-05-10: Review follow-up verified that SQL checkpoint saves
 already use a nested transaction savepoint for duplicate idempotency keys, so
 no whole-unit-of-work rollback fix was needed. Multi-step suspended checkpoints
 now fail explicitly during `resume_generation_orchestration` instead of
-silently finalising only the first externally supplied action result.
+silently finalizing only the first externally supplied action result.

Triage: [type:spelling]

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@docs/execplans/2-4-2-add-lang-graph-suspend-and-resume-orchestration.md`
around lines 689 - 718, Replace the three British spellings in the revision
notes to use Oxford -ize forms: change "serialised" to "serialized",
"deserialisation" to "deserialization", and "finalising" to "finalizing" in the
revision text (search for those exact words in the
docs/execplans/2-4-2-add-lang-graph-suspend-and-resume-orchestration.md revision
notes and update them).

♻️ Duplicate comments (2)

docs/developers-guide.md (1)

625-626: ⚠️ Potential issue | 🟡 Minor | ⚡ Quick win

Replace "independent from" with "independent of".

The correct en-GB collocation is "independent of", not "independent from".

Proposed fix

-  checkpoint, and resume ports that keep graph policy independent from storage,
+  checkpoint, and resume ports that keep graph policy independent of storage,

Triage: [type:grammar]

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@docs/developers-guide.md` around lines 625 - 626, Replace the phrase
"independent from" with "independent of" in the sentence containing "checkpoint,
and resume ports that keep graph policy independent from storage, queue, and
provider adapters" so the collocation uses en-GB "independent of" (i.e., change
"...independent from storage, queue, and provider adapters" to "...independent
of storage, queue, and provider adapters").

episodic/orchestration/langgraph.py (1)

485-486: ⚠️ Potential issue | 🟡 Minor | ⚡ Quick win

Guard against missing planner_result key in checkpoint payload.

Line 486 accesses payload["planner_result"] directly. A malformed checkpoint payload raises KeyError, but the docstring documents only TypeError and ValueError. Either catch KeyError and raise TypeError, or document KeyError in the docstring.

Proposed fix

     payload = checkpoint.payload
+    if "planner_result" not in payload:
+        msg = "checkpoint payload missing required key: planner_result"
+        raise TypeError(msg)
     planner_result = _planner_result_from_payload(payload["planner_result"])

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@episodic/orchestration/langgraph.py` around lines 485 - 486, The code
directly indexes payload["planner_result"] which can raise KeyError for
malformed checkpoint payloads; in the function handling checkpoint.payload
(referencing the local variable payload and the call to
_planner_result_from_payload), wrap the access in a try/except KeyError and
raise a TypeError (or convert to the documented exception) with a clear message
about the missing "planner_result" key so callers see a consistent error type,
or alternatively update the docstring to include KeyError if you prefer not to
change runtime behavior.

🤖 Prompt for all review comments with AI agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

Outside diff comments:
In `@docs/execplans/2-4-2-add-lang-graph-suspend-and-resume-orchestration.md`:
- Around line 689-718: Replace the three British spellings in the revision notes
to use Oxford -ize forms: change "serialised" to "serialized", "deserialisation"
to "deserialization", and "finalising" to "finalizing" in the revision text
(search for those exact words in the
docs/execplans/2-4-2-add-lang-graph-suspend-and-resume-orchestration.md revision
notes and update them).

---

Duplicate comments:
In `@docs/developers-guide.md`:
- Around line 625-626: Replace the phrase "independent from" with "independent
of" in the sentence containing "checkpoint, and resume ports that keep graph
policy independent from storage, queue, and provider adapters" so the
collocation uses en-GB "independent of" (i.e., change "...independent from
storage, queue, and provider adapters" to "...independent of storage, queue, and
provider adapters").

In `@episodic/orchestration/langgraph.py`:
- Around line 485-486: The code directly indexes payload["planner_result"] which
can raise KeyError for malformed checkpoint payloads; in the function handling
checkpoint.payload (referencing the local variable payload and the call to
_planner_result_from_payload), wrap the access in a try/except KeyError and
raise a TypeError (or convert to the documented exception) with a clear message
about the missing "planner_result" key so callers see a consistent error type,
or alternatively update the docstring to include KeyError if you prefer not to
change runtime behavior.

---

ℹ️ Review info

⚙️ Run configuration

Configuration used: Organization UI

Review profile: ASSERTIVE

Plan: Pro Plus

Run ID: 0e89e48e-0fd8-4268-96a6-edbd8a34b779

📥 Commits

Reviewing files that changed from the base of the PR and between 6a80044 and 374236c.

📒 Files selected for processing (4)

• docs/developers-guide.md
• docs/execplans/2-4-2-add-lang-graph-suspend-and-resume-orchestration.md
• episodic/orchestration/langgraph.py
• tests/test_generation_orchestration_langgraph.py

[coderabbitai]

Tip

For best results, initiate chat on the files or code changes.

🧩 Analysis chain

🏁 Script executed:

#!/bin/bash
# Check current state of build_workflow_step_idempotency_key in _dto.py
echo "=== _dto.py: build_workflow_step_idempotency_key and WorkflowStepIdentity ==="
rg -n "WorkflowStepIdentity\|build_workflow_step_idempotency_key\|PLR0913" episodic/orchestration/_dto.py -A 5 -B 2

echo ""
echo "=== test_orchestration_properties.py: test_step_idempotency_keys_are_deterministic ==="
rg -n "test_step_idempotency_keys_are_deterministic\|WorkflowStepIdentity\|step_identity_strategy\|_PropStepKeyInputs" tests/test_orchestration_properties.py -A 10 -B 2

echo ""
echo "=== langgraph.py: call site ==="
rg -n "build_workflow_step_idempotency_key" episodic/orchestration/langgraph.py -A 8 -B 2

echo ""
echo "=== __init__.py exports ==="
rg -n "WorkflowStepIdentity" episodic/orchestration/__init__.py

echo ""
echo "=== generation.py exports ==="
rg -n "WorkflowStepIdentity" episodic/orchestration/generation.py

Length of output: 776

---

The \| alternation didn't fire in rg's Rust regex engine. Let me re-check the two remaining files with correct syntax.

---

🏁 Script executed:

#!/bin/bash
echo "=== _dto.py: WorkflowStepIdentity ==="
rg -n "WorkflowStepIdentity" episodic/orchestration/_dto.py -A 5 -B 2

echo ""
echo "=== _dto.py: build_workflow_step_idempotency_key ==="
rg -n "build_workflow_step_idempotency_key" episodic/orchestration/_dto.py -A 10 -B 2

echo ""
echo "=== _dto.py: PLR0913 ==="
rg -n "PLR0913" episodic/orchestration/_dto.py

echo ""
echo "=== test_orchestration_properties.py: WorkflowStepIdentity ==="
rg -n "WorkflowStepIdentity" tests/test_orchestration_properties.py -A 5 -B 2

echo ""
echo "=== test_orchestration_properties.py: test_step_idempotency_keys_are_deterministic ==="
rg -n "test_step_idempotency_keys_are_deterministic" tests/test_orchestration_properties.py -A 20 -B 3

Length of output: 2933

---

@leynos: Both CodeScene violations have been fully resolved. Here is a summary of the current state:

Location	Violation	Status
episodic/orchestration/_dto.py → build_workflow_step_idempotency_key	5 args (max 4)	✅ Resolved — now 2 params (step: WorkflowStepIdentity, attempt: int); # noqa: PLR0913 removed
tests/test_orchestration_properties.py → test_step_idempotency_keys_are_deterministic	5 args (max 4)	✅ Resolved — now 2 params (inputs: _PropStepKeyInputs, attempt: int)

The supporting changes are also all in place:

• WorkflowStepIdentity is defined in episodic/orchestration/_dto.py as a frozen, slotted dataclass.
• The call site in episodic/orchestration/langgraph.py passes a single identity argument.
• WorkflowStepIdentity is exported from both episodic/orchestration/__init__.py and episodic/orchestration/generation.py.

No further work is required for these checks.

[coderabbitai]

Actionable comments posted: 6

♻️ Duplicate comments (3)

episodic/canonical/storage/workflow_checkpoints.py (1)

11-11: ⚠️ Potential issue | 🟡 Minor | ⚡ Quick win

Remove the redundant future-annotations import in this new module.

Delete the import at Line 11 to align with the Python 3.14 baseline.

♻️ Suggested patch

-from __future__ import annotations
-
 import typing as typ

As per coding guidelines, "Do not add from __future__ import annotations in new or modified files; the project baseline is Python >=3.14 where annotation evaluation is deferred by default".

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@episodic/canonical/storage/workflow_checkpoints.py` at line 11, Remove the
redundant future import statement "from __future__ import annotations" from the
top of the module (the import shown in the diff) so the file relies on the
project Python >=3.14 baseline; simply delete that import line and ensure no
other references depend on it (no other changes required).

docs/episodic-podcast-generation-system-design.md (1)

553-600: ⚠️ Potential issue | 🟡 Minor | ⚡ Quick win

Add screen-reader descriptions before both new Mermaid sequence diagrams.

Insert a short Screen reader description: paragraph before the diagram at
Line 553 and before the diagram at Line 577 so assistive technologies receive
context before parsing the diagram blocks.

♿ Suggested patch

+Screen reader description: Suspend path for generation where planning completes,
+the checkpoint is persisted or reused, and the workflow returns a suspended
+result before side-effecting execution.
+
 ```mermaid
 sequenceDiagram
@@

@@
+Screen reader description: Resume path where checkpoint lookup rejects unknown
+IDs, known checkpoints resume through TaskResumePort, and the final generation
+result is built before marking the checkpoint resumed.
+

sequenceDiagram

Loading

Triage: [type:docstyle]

As per coding guidelines, "Add screen reader descriptions before complex diagrams or code blocks in documentation".

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@docs/episodic-podcast-generation-system-design.md` around lines 553 - 600,
Add a short "Screen reader description:" paragraph immediately before each
mermaid sequenceDiagram block shown (the one labeled
StructuredPlanningOrchestrator/GenerationGraph and the one labeled
GenerationOrchestrationLayer/CheckpointPort/TaskResumePort) so assistive tech
gets context; the paragraph should be one or two concise sentences describing
the flow (e.g., "Screen reader description: Suspend flow that persists or reuses
a checkpoint before side-effecting execution." for the first diagram and "Screen
reader description: Resume path where checkpoint lookup rejects unknown IDs,
known checkpoints resume through TaskResumePort and the final generation result
is built before marking the checkpoint resumed." for the second), inserted
directly above the opening ```mermaid line for each diagram.

tests/canonical_storage/test_workflow_checkpoints.py (1)

3-3: ⚠️ Potential issue | 🟡 Minor | ⚡ Quick win

Remove the redundant future-annotations import in this new test module.

Delete Line 3 to match the repository Python 3.14 baseline.

♻️ Suggested patch

-from __future__ import annotations
-
 import typing as typ

As per coding guidelines, "Do not add from __future__ import annotations in new or modified files; the project baseline is Python >=3.14 where annotation evaluation is deferred by default".

🤖 Prompt for AI Agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@tests/canonical_storage/test_workflow_checkpoints.py` at line 3, Delete the
redundant top-level line "from __future__ import annotations" in the new test
module (the import shown in the diff); remove that import statement so the file
no longer includes it and save the file.

🤖 Prompt for all review comments with AI agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

Inline comments:
In `@alembic/versions/20260508_000008_add_workflow_checkpoints.py`:
- Around line 36-41: The migration defines updated_at with
server_default=sa.func.now() but lacks a DB-side update on row modification; if
rows may be updated outside the ORM add a PostgreSQL trigger in this migration:
create a plpgsql function (e.g. update_updated_at_column) that sets
NEW.updated_at = NOW() and create a BEFORE UPDATE trigger (e.g.
update_workflow_checkpoints_updated_at) on the workflow_checkpoints table via
op.execute in the upgrade, and in the downgrade drop the trigger and the
function via op.execute; reference the updated_at column and the table name
workflow_checkpoints and ensure the function/trigger names match between upgrade
and downgrade.

In `@docs/execplans/2-4-2-add-lang-graph-suspend-and-resume-orchestration.md`:
- Around line 465-467: The ExecPlan currently references reserve_or_get(...);
update the document to use the final API name save_or_reuse(...) everywhere in
the ExecPlan so it matches the implemented port contract and avoids
confusion—specifically replace occurrences of reserve_or_get with save_or_reuse
and ensure the surrounding text still describes the method as atomic via an
idempotency_key uniqueness constraint (preserve mentions of atomicity and
idempotency_key).

In `@episodic/canonical/storage/models.py`:
- Around line 104-108: The status column on the model is a plain String and
needs DB-level validation; define a domain enum (e.g., WorkflowCheckpointStatus
with values "suspended" and "resumed") in episodic/canonical/domain.py and
change the model's status mapped_column to use sa.Enum(WorkflowCheckpointStatus,
name="workflow_checkpoint_status") (or an equivalent CHECK-backed Enum) so only
allowed values are persisted; update any imports and adjust the mapped attribute
(status: orm.Mapped[str]) to reference the enum type to ensure SQLAlchemy and
the DB enforce the valid statuses.
- Around line 79-80: Add a NumPy-style docstring to the WorkflowCheckpointRecord
class that mirrors the structure used by other models (e.g.,
SeriesProfileRecord) and includes an "Attributes" section documenting each
SQLAlchemy column on the model: list each attribute name (e.g., id, workflow_id,
checkpoint_data, created_at, updated_at or the actual column names in
WorkflowCheckpointRecord), its type (e.g., Integer, String, JSON, DateTime), and
any constraints (primary key, nullable, unique, foreign key, default) plus a
one-line purpose for each; ensure the docstring is placed immediately under the
class definition and follows the same formatting and phrasing conventions used
in other model docstrings in the file.

In `@episodic/orchestration/langgraph.py`:
- Around line 167-173: The code currently accesses step["required_inputs"]
directly which raises KeyError; instead call _require_field(step,
"required_inputs") to preserve the TypeError contract—update the required_inputs
construction (the tuple(...) comprehension that uses typ.cast("list[object]",
...)) to use _require_field(step, "required_inputs") in place of
step["required_inputs"], leaving the cast and tuple generation intact so missing
fields raise TypeError as expected.

In `@tests/test_generation_orchestration_snapshots.py`:
- Around line 20-23: The import line currently brings in unused symbols
(_action_result_from_payload, _plan_from_payload, _plan_to_payload,
_planner_result_from_payload); update the import from
episodic.orchestration.langgraph to only import the two used functions
_action_result_to_payload and _planner_result_to_payload so the module no longer
imports unused names.

---

Duplicate comments:
In `@docs/episodic-podcast-generation-system-design.md`:
- Around line 553-600: Add a short "Screen reader description:" paragraph
immediately before each mermaid sequenceDiagram block shown (the one labeled
StructuredPlanningOrchestrator/GenerationGraph and the one labeled
GenerationOrchestrationLayer/CheckpointPort/TaskResumePort) so assistive tech
gets context; the paragraph should be one or two concise sentences describing
the flow (e.g., "Screen reader description: Suspend flow that persists or reuses
a checkpoint before side-effecting execution." for the first diagram and "Screen
reader description: Resume path where checkpoint lookup rejects unknown IDs,
known checkpoints resume through TaskResumePort and the final generation result
is built before marking the checkpoint resumed." for the second), inserted
directly above the opening ```mermaid line for each diagram.

In `@episodic/canonical/storage/workflow_checkpoints.py`:
- Line 11: Remove the redundant future import statement "from __future__ import
annotations" from the top of the module (the import shown in the diff) so the
file relies on the project Python >=3.14 baseline; simply delete that import
line and ensure no other references depend on it (no other changes required).

In `@tests/canonical_storage/test_workflow_checkpoints.py`:
- Line 3: Delete the redundant top-level line "from __future__ import
annotations" in the new test module (the import shown in the diff); remove that
import statement so the file no longer includes it and save the file.

🪄 Autofix (Beta)

Fix all unresolved CodeRabbit comments on this PR:

• Push a commit to this branch (recommended)
• Create a new PR with the fixes

---

ℹ️ Review info

⚙️ Run configuration

Configuration used: Organization UI

Review profile: ASSERTIVE

Plan: Pro Plus

Run ID: 98506b4c-dbc7-4765-8472-f18918d30498

📥 Commits

Reviewing files that changed from the base of the PR and between 34093b8 and 57d5761.

📒 Files selected for processing (25)

• alembic/versions/20260508_000008_add_workflow_checkpoints.py
• docs/adr/adr-007-durable-generation-checkpoints.md
• docs/developers-guide.md
• docs/episodic-podcast-generation-system-design.md
• docs/execplans/2-4-2-add-lang-graph-suspend-and-resume-orchestration.md
• docs/roadmap.md
• docs/users-guide.md
• episodic/canonical/storage/__init__.py
• episodic/canonical/storage/models.py
• episodic/canonical/storage/uow.py
• episodic/canonical/storage/workflow_checkpoints.py
• episodic/orchestration/__init__.py
• episodic/orchestration/_dto.py
• episodic/orchestration/_protocols.py
• episodic/orchestration/checkpoints.py
• episodic/orchestration/generation.py
• episodic/orchestration/langgraph.py
• tests/__snapshots__/test_generation_orchestration_snapshots.ambr
• tests/canonical_storage/test_workflow_checkpoints.py
• tests/features/generation_orchestration.feature
• tests/steps/test_generation_orchestration_steps.py
• tests/test_generation_orchestration_langgraph.py
• tests/test_generation_orchestration_snapshots.py
• tests/test_orchestration_planner.py
• tests/test_orchestration_properties.py

[leynos]

@coderabbitai Have the following now been resolved?

Annul any requirements that violate the en-GB-oxendict spelling (-ize / -yse / -our) conventions (for example a request to replace "normalize" with "normalise" or "artefact" with "artifact"), or where the requirement unnecessarily increases cyclomatic complexity.

Please address the comments from this code review:

## Overall Comments
- In `_planner_result_from_payload`, `planner_payload["usage"]` is always passed into `_usage_from_payload`, but `_planner_result_to_payload` explicitly allows `usage` to be `None`; this will raise a `TypeError` on resume for planner results without usage, so `None` should be handled symmetrically in the deserialization path.
- `resume_generation_orchestration` currently reads the checkpoint and resumes via `TaskResumePort` but never updates the stored checkpoint status (e.g. to `resumed`) through `CheckpointPort`, which makes it hard to distinguish active vs. completed checkpoints and may complicate future cleanup or monitoring; consider adding an explicit status transition when resume completes.

## Individual Comments

### Comment 1
<location path="episodic/orchestration/langgraph.py" line_range="149-154" />
<code_context>
+    }
+
+
+def _planner_result_from_payload(payload: object) -> dto.PlannerResult:
+    """Return a PlannerResult from a checkpoint payload."""
+    planner_payload = _as_object_payload(payload, "planner_result")
+    return dto.PlannerResult(
+        plan=_plan_from_payload(planner_payload["plan"]),
+        usage=_usage_from_payload(planner_payload["usage"]),
+        model=_required_string(planner_payload, "model"),
+        provider_response_id=_required_string(planner_payload, "provider_response_id"),
</code_context>
<issue_to_address>
**issue (bug_risk):** PlannerResult deserialization assumes a non-null `usage`, but the serialization can emit `usage=None`.

`_planner_result_to_payload` can serialize `usage` as `None` via `_usage_to_payload`, but `_planner_result_from_payload` always calls `_usage_from_payload(planner_payload["usage"])`, which will raise a `TypeError` when the stored value is `None` (since `_as_object_payload` expects a `dict`). To match `_action_result_from_payload` and correctly support optional `usage`, you could use:

```python
usage=(
    None
    if planner_payload.get("usage") is None
    else _usage_from_payload(planner_payload["usage"])
),
```
</issue_to_address>

### Comment 2
<location path="episodic/canonical/storage/workflow_checkpoints.py" line_range="32-35" />
<code_context>
+    )
+
+
+class SqlAlchemyWorkflowCheckpointStore:
+    """SQLAlchemy implementation of the orchestration CheckpointPort."""
+
+    def __init__(self, session: AsyncSession) -> None:
+        self._session = session
+
</code_context>
<issue_to_address>
**issue (bug_risk):** Using `AsyncSession` directly in the type annotation will raise a `NameError` at runtime under `TYPE_CHECKING`-guarded import.

Because `AsyncSession` is only imported under `if typ.TYPE_CHECKING`, it doesn’t exist at runtime, so evaluating the annotation on `__init__` will raise a `NameError`.

To fix this, either:
- Use a string annotation: `def __init__(self, session: "AsyncSession") -> None:`
- Or import `AsyncSession` unconditionally and keep the current annotation.

Both options preserve type checking without breaking at runtime.
</issue_to_address>

### Comment 3
<location path="tests/test_orchestration_properties.py" line_range="133-138" />
<code_context>
+)
+
+
+@given(
+    inputs=_step_key_inputs_strategy,
+    attempt=st.integers(min_value=0, max_value=100),
+)
+@settings(max_examples=50)
+def test_step_idempotency_keys_are_deterministic(
+    inputs: _PropStepKeyInputs,
+    attempt: int,
</code_context>
<issue_to_address>
**suggestion (testing):** Complement the property test with a unit test for negative `attempt` values in `build_workflow_step_idempotency_key`.

The property test covers determinism and positive attempts, but not the branch where `attempt < 0` raises `ValueError("attempt must be greater than or equal to zero.")`. Please add a simple unit test (e.g. `with pytest.raises(ValueError): build_workflow_step_idempotency_key(step, attempt=-1)`) to cover this validation and guard against regressions.

Suggested implementation:

```python
        alphabet=string.ascii_letters + string.digits + "_",
    ),
    action_id=st.text(
        min_size=1,
        max_size=32,
        alphabet=string.ascii_letters + string.digits + "-",
    ),
)


def test_step_idempotency_key_negative_attempt_raises_value_error() -> None:
    step = WorkflowStepIdentity(
        workflow_id="workflow-id",
        workflow_type="workflow-type",
        step_name="step-name",
        action_id="action-id",
    )

    with pytest.raises(ValueError):
        build_workflow_step_idempotency_key(step, attempt=-1)


@given(

```

1. At the top of `tests/test_orchestration_properties.py`, ensure `pytest` is imported:
   `import pytest`.
2. If this file is using a specific fixture or factory for `WorkflowStepIdentity` in other tests, you may prefer to construct `step` using that helper instead of hard-coded strings for consistency.
</issue_to_address>

### Comment 4
<location path="tests/canonical_storage/test_workflow_checkpoints.py" line_range="32-33" />
<code_context>
+    )
+
+
+@pytest.mark.asyncio
+async def test_checkpoint_store_persists_across_unit_of_work(
+    session_factory: object,
+) -> None:
</code_context>
<issue_to_address>
**suggestion (testing):** Consider adding tests for `get`/`get_by_idempotency_key` when no checkpoint exists.

It would be useful to also assert the "not found" behaviour of `SqlAlchemyWorkflowCheckpointStore`:

- `get(non_existent_uuid)` returns `None`
- `get_by_idempotency_key(non_existent_key)` returns `None`

You can either extend the existing tests or add a small dedicated test to cover these cases, so the expected behaviour for missing checkpoints is explicit.

Suggested implementation:

```python
@pytest.mark.asyncio
async def test_checkpoint_store_persists_across_unit_of_work(
    session_factory: object,
) -> None:
    """Checkpoint records should survive fresh unit-of-work instances."""
    factory = typ.cast("async_sessionmaker[AsyncSession]", session_factory)
    checkpoint = _checkpoint()

    async with SqlAlchemyUnitOfWork(factory) as uow:
        stored = await uow.workflow_checkpoints.save(checkpoint)
        await uow.commit()

    async with SqlAlchemyUnitOfWork(factory) as uow:
        fetched = await uow.workflow_checkpoints.get(stored.checkpoint_id)

    assert fetched is not None
    assert fetched.checkpoint_id == stored.checkpoint_id
    assert fetched.idempotency_key == stored.idempotency_key


@pytest.mark.asyncio
async def test_checkpoint_store_get_returns_none_for_missing_checkpoint(
    session_factory: object,
) -> None:
    """`get` should return None when the checkpoint does not exist."""
    factory = typ.cast("async_sessionmaker[AsyncSession]", session_factory)
    missing_checkpoint_id = uuid.uuid4()

    async with SqlAlchemyUnitOfWork(factory) as uow:
        result = await uow.workflow_checkpoints.get(missing_checkpoint_id)

    assert result is None


@pytest.mark.asyncio
async def test_checkpoint_store_get_by_idempotency_key_returns_none_for_missing_checkpoint(
    session_factory: object,
) -> None:
    """`get_by_idempotency_key` should return None when the checkpoint does not exist."""
    factory = typ.cast("async_sessionmaker[AsyncSession]", session_factory)
    missing_idempotency_key = "non-existent-idempotency-key"

    async with SqlAlchemyUnitOfWork(factory) as uow:
        result = await uow.workflow_checkpoints.get_by_idempotency_key(
            missing_idempotency_key
        )

    assert result is None

```

If `uuid` is not already imported in `tests/canonical_storage/test_workflow_checkpoints.py`, add:

- `import uuid`

to the import section at the top of the file.
</issue_to_address>

### Comment 5
<location path="episodic/orchestration/langgraph.py" line_range="47" />
<code_context>
+    }
+
+
+def _as_object_payload(payload: object, field_name: str) -> dict[str, object]:
+    """Return payload as a string-keyed object."""
+    if not isinstance(payload, dict):
</code_context>
<issue_to_address>
**issue (complexity):** Consider refactoring the new helper functions, suspend logic, and graph construction to centralise shared logic and separate concerns, reducing boilerplate and making the orchestration flow easier to follow.

You can keep the new functionality but trim a fair bit of incidental complexity with a few small refactors.

### 1. Collapse `_required_*` helpers into a single generic getter

You have three very similar helpers (`_as_object_payload`, `_required_int`, `_required_string`). You can reduce boilerplate and make future changes easier by centralising the type check logic:

```python
def _require_field[T](
    payload: dict[str, object],
    field_name: str,
    expected_type: type[T],
    *,
    context: str,
) -> T:
    try:
        value = payload[field_name]
    except KeyError as exc:
        msg = f"checkpoint {context} missing required field: {field_name}"
        raise KeyError(msg) from exc

    if not isinstance(value, expected_type):
        msg = f"checkpoint {context} field {field_name} must be a {expected_type.__name__}"
        raise TypeError(msg)
    return typ.cast(T, value)
```

Then the specific helpers become thin wrappers (or can be removed entirely):

```python
def _as_object_payload(payload: object, context: str) -> dict[str, object]:
    if not isinstance(payload, dict):
        msg = f"checkpoint {context} payload must be an object."
        raise TypeError(msg)
    return typ.cast("dict[str, object]", payload)

def _required_int(payload: dict[str, object], field_name: str, *, context: str) -> int:
    return _require_field(payload, field_name, int, context=context)

def _required_string(payload: dict[str, object], field_name: str, *, context: str) -> str:
    return _require_field(payload, field_name, str, context=context)
```

Usage then becomes more self‑describing and avoids repeated error messages:

```python
plan_payload = _as_object_payload(payload, "plan")
plan_version = _required_string(plan_payload, "plan_version", context="plan")

usage_payload = _as_object_payload(payload, "usage")
input_tokens = _required_int(usage_payload, "input_tokens", context="usage")
```

This keeps behaviour identical but centralises the validation logic.

### 2. Extract checkpoint payload and identity construction from `_suspend_execute_node`

`_suspend_execute_node` currently mixes state validation, identity/idempotency construction, checkpoint I/O, and payload building. You can make it much easier to scan by extracting the mechanical bits:

```python
def _build_execute_step_identity(
    *,
    request: dto.GenerationOrchestrationRequest,
    action: dto.PlannedAction,
    workflow_type: str,
) -> dto.WorkflowStepIdentity:
    return dto.WorkflowStepIdentity(
        workflow_id=request.correlation_id,
        workflow_type=workflow_type,
        step_name="execute",
        action_id=action.action_id,
    )


def _build_checkpoint_payload(
    *,
    request: dto.GenerationOrchestrationRequest,
    planner_result: dto.PlannerResult,
) -> dict[str, object]:
    return {
        "request": {
            "correlation_id": request.correlation_id,
            "script_tei_xml": request.script_tei_xml,
            "template_structure": request.template_structure,
        },
        "planner_result": _planner_result_to_payload(planner_result),
    }
```

Then `_suspend_execute_node` focuses on orchestration flow rather than wiring:

```python
async def _suspend_execute_node(
    state: GenerationGraphState,
    *,
    checkpoint_port: protocols.CheckpointPort,
    workflow_type: str = "generation_orchestration",
) -> dict[str, dto.SuspendedWorkflowResult]:
    request = state.request
    if request is None:
        raise ValueError("missing required state value: request")

    planner_result = state.planner_result
    if planner_result is None:
        raise ValueError("missing required state value: planner_result")
    if not planner_result.plan.steps:
        raise ValueError("cannot suspend a workflow with no planned steps")

    action = planner_result.plan.steps[0]

    identity = _build_execute_step_identity(
        request=request,
        action=action,
        workflow_type=workflow_type,
    )
    idempotency_key = dto.build_workflow_step_idempotency_key(identity)

    existing = await checkpoint_port.get_by_idempotency_key(idempotency_key)
    if existing is None:
        payload = _build_checkpoint_payload(
            request=request,
            planner_result=planner_result,
        )
        existing = await checkpoint_port.save(
            dto.WorkflowCheckpoint(
                checkpoint_id=str(uuid.uuid4()),
                workflow_id=identity.workflow_id,
                workflow_type=identity.workflow_type,
                step_name=identity.step_name,
                idempotency_key=idempotency_key,
                payload=payload,
            )
        )

    suspended_result = dto.SuspendedWorkflowResult(
        checkpoint_id=existing.checkpoint_id,
        workflow_id=existing.workflow_id,
        step_name=existing.step_name,
        idempotency_key=existing.idempotency_key,
    )
    return {"suspended_result": suspended_result}
```

Same behaviour, but the orchestration path is clearer and most of the low‑level details are encapsulated.

### 3. Simplify graph construction branch in `build_generation_orchestration_graph`

Instead of inlining two slightly different graph shapes, pick the execute node function and edge target up front, then build the graph once:

```python
def build_generation_orchestration_graph(
    *,
    planner: protocols.PlannerPort,
    tool_executor: protocols.ToolExecutorPort,
    checkpoint_port: protocols.CheckpointPort | None = None,
) -> CompiledStateGraph[GenerationGraphState, None, GenerationGraphState, GenerationGraphState]:
    graph = StateGraph(GenerationGraphState)

    async def _run_plan_node(state: GenerationGraphState) -> dict[str, dto.PlannerResult]:
        return await _plan_node(state, planner=planner)

    async def _run_execute_node(
        state: GenerationGraphState,
    ) -> dict[str, tuple[dto.ActionExecutionResult, ...]]:
        return await _execute_node(state, tool_executor=tool_executor)

    if checkpoint_port is None:
        execute_node = _run_execute_node
        execute_target = "finish"
    else:

        async def _run_suspend_execute_node(
            state: GenerationGraphState,
        ) -> dict[str, dto.SuspendedWorkflowResult]:
            return await _suspend_execute_node(
                state,
                checkpoint_port=checkpoint_port,
            )

        execute_node = _run_suspend_execute_node
        execute_target = END

    graph.add_node("plan", _run_plan_node)
    graph.add_node("execute", execute_node)
    graph.add_node("finish", _finish_node)

    graph.add_edge(START, "plan")
    graph.add_edge("plan", "execute")
    graph.add_edge("execute", execute_target)
    graph.add_edge("finish", END)

    return graph.compile()
```

This removes duplicated “plan → execute” wiring and makes the two modes differ only in `execute_node`/`execute_target`, which matches the conceptual difference (“execute now” vs “suspend before execute”).
</issue_to_address>

### Comment 6
<location path="docs/execplans/2-4-2-add-lang-graph-suspend-and-resume-orchestration.md" line_range="68" />
<code_context>
+- Add pytest unit tests for every new code unit, pytest-bdd behavioural tests
+  for the observable suspend/resume workflow, and Hypothesis property tests for
+  idempotency-key and state-transition invariants.
+- Update `docs/episodic-podcast-generation-system-design.md`,
+  `docs/users-guide.md`, and `docs/developers-guide.md` for the implemented
+  behaviour. Add or update an ADR in `docs/adr/` for the durable design
</code_context>
<issue_to_address>
**issue (review_instructions):** Expand the acronym "ADR" on first use, such as "Architecture Decision Record (ADR)".

In the following line of this bullet you refer to "an ADR" without having expanded the term earlier in the document. To comply with the acronym guideline, please update the first mention to something like "Add or update an Architecture Decision Record (ADR) in `docs/adr/`".

<details>
<summary>Review instructions:</summary>

**Path patterns:** `**/*.md`

**Instructions:**
Define uncommon acronyms on first use.

</details>
</issue_to_address>

[leynos]

@coderabbitai Please suggest a fix for this issue and supply a prompt for an AI coding agent to enable it to apply the fix. Include the file and symbol names indicated in the issue at the head of your response.

episodic/orchestration/langgraph.py

Comment on lines +388 to +460

async def _suspend_execute_node(
    state: GenerationGraphState,
    *,
    checkpoint_port: protocols.CheckpointPort,
    workflow_type: str = "generation_orchestration",
) -> dict[str, dto.SuspendedWorkflowResult]:
    """Persist or reuse a checkpoint before executing the first action."""
    request = state.request
    if request is None:
        msg = "missing required state value: request"
        raise ValueError(msg)
    planner_result = state.planner_result
    if planner_result is None:
        msg = "missing required state value: planner_result"
        raise ValueError(msg)
    if len(planner_result.plan.steps) != 1:
        msg = "cannot suspend a workflow with no planned steps"
        if planner_result.plan.steps:
            msg = (
                "suspend_generation_orchestration currently supports exactly "
                "one planned step per suspended checkpoint."
            )
        raise ValueError(msg)

    action = planner_result.plan.steps[0]
    identity = _build_execute_step_identity(
        request=request,
        action=action,
        workflow_type=workflow_type,
    )
    idempotency_key = dto.build_workflow_step_idempotency_key(
        identity,
    )
    _log_event(
        "debug",
        "generation_graph.suspend_execute_node.start",
        correlation_id=request.correlation_id,
        workflow_id=identity.workflow_id,
        workflow_type=identity.workflow_type,
        step_name=identity.step_name,
        action_id=identity.action_id,
        idempotency_key=idempotency_key,
    )
    fresh_id = str(uuid.uuid4())
    existing = await checkpoint_port.save_or_reuse(
        dto.WorkflowCheckpoint(
            checkpoint_id=fresh_id,
            workflow_id=identity.workflow_id,
            workflow_type=identity.workflow_type,
            step_name=identity.step_name,
            idempotency_key=idempotency_key,
            payload=_build_checkpoint_payload(
                request=request,
                planner_result=planner_result,
            ),
        )
    )
    reused_checkpoint = existing.checkpoint_id != fresh_id
    _log_event(
        "debug",
        "generation_graph.suspend_execute_node.finish",
        correlation_id=request.correlation_id,
        checkpoint_id=existing.checkpoint_id,
        idempotency_key=existing.idempotency_key,
        reused_checkpoint=reused_checkpoint,
    )
    suspended_result = dto.SuspendedWorkflowResult(
        checkpoint_id=existing.checkpoint_id,
        workflow_id=existing.workflow_id,
        step_name=existing.step_name,
        idempotency_key=existing.idempotency_key,
    )
    return {"suspended_result": suspended_result}

❌ New issue: Large Method
_suspend_execute_node has 72 lines, threshold = 70

[coderabbitai]

Tip

For best results, initiate chat on the files or code changes.

🧩 Analysis chain

🏁 Script executed:

#!/bin/bash
echo "=== Unit Architecture: save_or_reuse presence ==="
grep -n "save_or_reuse\|def save" episodic/orchestration/_protocols.py episodic/orchestration/checkpoints.py episodic/canonical/storage/workflow_checkpoints.py

echo ""
echo "=== Performance: _suspend_execute_node - pre-save get_by_idempotency_key ==="
grep -n "get_by_idempotency_key\|save_or_reuse\|save\b\|fresh_id" episodic/orchestration/langgraph.py | head -30

echo ""
echo "=== Observability: _log_event in SqlAlchemyWorkflowCheckpointStore ==="
grep -n "_log_event\|log\b\|structlog\|logging" episodic/canonical/storage/workflow_checkpoints.py | head -20

echo ""
echo "=== Testing (Unit & Behavioural): mark_resumed unknown id tests ==="
grep -n "mark_resumed_raises\|mark_resumed.*unknown\|unknown.*mark_resumed\|unknown_checkpoint_id\|raises_for_unknown" tests/test_generation_orchestration_langgraph.py tests/canonical_storage/test_workflow_checkpoints.py 2>/dev/null

echo ""
echo "=== Testing: invalid payload deserialisation test ==="
grep -n "invalid_payload\|invalid.*payload\|payload.*invalid\|raises_on_invalid\|deserialisation" tests/test_generation_orchestration_langgraph.py 2>/dev/null

echo ""
echo "=== Testing (Property): concurrent/Hypothesis checkpoint tests ==="
grep -n "concurrent\|gather\|asyncio\|status.*transition\|first.write\|save_or_reuse" tests/test_orchestration_properties.py | head -20

echo ""
echo "=== Existing GitHub issues ==="
gh issue list --repo leynos/episodic --state open --limit 50 --json number,title | jq '.[] | {number, title}'

echo ""
echo "=== Large Method check: _suspend_execute_node ==="
sed -n '/^async def _suspend_execute_node/,/^async def \|^def \|^class /p' episodic/orchestration/langgraph.py | head -10

echo ""
echo "=== _validate_suspend_preconditions helper ==="
grep -n "_validate_suspend_preconditions" episodic/orchestration/langgraph.py | head -5

Length of output: 4813

---

@leynos: All four in-scope checks are now resolved. Here is the full status:

Check-by-check verdict

Check	Status	Evidence
Unit Architecture ❌	✅ Resolved	save_or_reuse is present in _protocols.py, checkpoints.py, and workflow_checkpoints.py; no def save survives anywhere
Performance And Resource Use ⚠️	✅ Resolved	_suspend_execute_node calls save_or_reuse unconditionally via a fresh_id; the pre-save get_by_idempotency_key is gone
Observability ⚠️	✅ Resolved	_log_event imported and called at five points in SqlAlchemyWorkflowCheckpointStore (get, get_by_idempotency_key, save_or_reuse persisted path, save_or_reuse conflict path, mark_resumed)
Testing (Unit And Behavioural) ⚠️	✅ Resolved	All three missing tests are present: test_resume_generation_orchestration_raises_on_invalid_payload (line 317), test_in_memory_checkpoint_store_mark_resumed_raises_for_unknown_id (line 374), and test_checkpoint_store_mark_resumed_raises_for_unknown_checkpoint_id (line 117)
Testing (Property / Proof) ⚠️	🔖 Out of scope	No Hypothesis concurrent or status-transition property tests exist; tracked in Issue #72
Concurrency And State ⚠️	🔖 Out of scope	Tracked in Issue #91
Testing (Overall) ❓	—	Inconclusive; nothing actionable
Security And Privacy ❓	—	Inconclusive; nothing actionable

No further work is required on this PR for the in-scope items. Extend Issue #72 to cover concurrent save_or_reuse Hypothesis tests and status-transition sequences when that work is scheduled.

[coderabbitai]

Actionable comments posted: 2

🤖 Prompt for all review comments with AI agents

Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

Inline comments:
In `@docs/adr/adr-007-durable-generation-checkpoints.md`:
- Line 1: Update the document title to expand the ADR acronym on first use:
replace the heading "# ADR-007: Durable generation checkpoints" with
"Architecture Decision Record (ADR)-007: Durable generation checkpoints" (or
equivalent) so the acronym ADR is expanded at first occurrence in the title.

In `@episodic/orchestration/langgraph.py`:
- Around line 157-166: Enum construction for PlannedAction fields
(dto.ActionKind and dto.ModelTier) can raise ValueError on invalid members which
bypasses the intended deserialisation TypeError path; update the deserialisation
logic around PlannedAction creation in langgraph.py (where _required_string is
used to build dto.ActionKind and dto.ModelTier) to catch ValueError from the
enum constructors and re-raise a TypeError with the same context, and apply the
same change to the other enum parsing site that constructs
dto.ActionKind/dto.ModelTier later in the file so all malformed-enum payloads
consistently produce TypeError.

🪄 Autofix (Beta)

Fix all unresolved CodeRabbit comments on this PR:

• Push a commit to this branch (recommended)
• Create a new PR with the fixes

---

ℹ️ Review info

⚙️ Run configuration

Configuration used: Organization UI

Review profile: ASSERTIVE

Plan: Pro Plus

Run ID: fc2ab477-3adf-4ac0-95f6-3d55d4c18ac9

📥 Commits

Reviewing files that changed from the base of the PR and between 57b9c1b and 20c0abc.

📒 Files selected for processing (2)

• docs/adr/adr-007-durable-generation-checkpoints.md
• episodic/orchestration/langgraph.py