Collapsing the tprl_content_generation/ workflow layer

1 — The shape of the problem (before)

When a PipeLLM operator inside WfPipeRouter needed to generate text, the call path went through three execution levels: router workflow, content-generation workflow, activity.

WfPipeRouter  (parent workflow)
    │
    ├─ pipe_llm._live_run_operator_pipe(...)
    │     └─ get_content_generator()           ◄── returns ContentGeneratorChild
    │     └─ content_generator.make_llm_text(...)
    │            │
    │            ▼
    │     ContentGeneratorChild.make_llm_text(...)
    │            │
    │            └─ WorkflowExecutorFactory[...].execute_child_workflow(WfMakeLLMText, ...)
    │                        │
    │                        ▼
    │              ┌───────────────────────────────────────────────┐
    │              │  WfMakeLLMText  (child workflow)              │
    │              │                                               │
    │              │   workflow.start_activity(                    │
    │              │     activity=act_llm_gen_text,                │
    │              │     arg=llm_assignment,                       │
    │              │     start_to_close_timeout=...,               │
    │              │     retry_policy=...,                         │
    │              │     task_queue=inference_task_queue,  ◄── LLM-text only
    │              │   )                                           │
    │              │                                               │
    │              │   try / except ActivityError →                │
    │              │     TemporalError.from_app_error(...)         │
    │              └───────────────────────────────────────────────┘
    │                        │
    │                        ▼
    │                act_llm_gen_text  (activity)
    │                        │
    └───── result ◄──────────┘

Every WfMake* workflow had the exact same shape:

@workflow.defn(name="wf_…")
class WfMakeXxx(WorkflowClass[XxxAssignment, ResultType]):
    @workflow.run
    async def run(self, workflow_arg: XxxAssignment) -> ResultType:
        worker_config = get_config().temporal.worker_config
        try:
            result = await workflow.start_activity(
                activity=act_xxx,
                arg=workflow_arg,
                start_to_close_timeout=worker_config.workflow_execution_timeout,
                retry_policy=worker_config.retry_policy,
                # task_queue=worker_config.inference_task_queue  ← LLM-text only
            )
        except ActivityError as exc:
            if isinstance(exc.cause, ApplicationError):
                raise TemporalError.from_app_error(exc=exc.cause) from exc
            raise
        return result

Seven copies of the same template, one per content-generation activity:

There was no remaining workflow body that did anything between activities. The previous text-then-object stack — the one nontrivial case in the v1 plan — had already been removed in commit 16b775b8. Every surviving WfMake* was just 1 activity = 1 result.

2 — The Temporal history cost

A single make_llm_text from inside WfPipeRouter produced this event chain in the parent's history:

[parent: WfPipeRouter]
  StartChildWorkflowExecutionInitiated
  ChildWorkflowExecutionStarted        ◄── child framing
  ChildWorkflowExecutionCompleted/Failed

[child: WfMakeLLMText]
  WorkflowExecutionStarted
  WorkflowTaskScheduled/Started/Completed
  ActivityTaskScheduled
  ActivityTaskStarted
  ActivityTaskCompleted/Failed
  WorkflowTaskScheduled/Started/Completed
  WorkflowExecutionCompleted

≈ 12 events per LLM call

After the collapse, the same call inside the parent is just:

[parent: WfPipeRouter]
  ActivityTaskScheduled
  ActivityTaskStarted
  ActivityTaskCompleted/Failed

3 events per LLM call

For a long PipeSequence of N LLM steps, the history goes from "N child-workflow scheduling pairs in the parent + N small child histories" to "N activity scheduling pairs in the parent". Per-LLM-call durability is preserved — the activity boundary is unchanged, so the retry semantics and the durable-checkpoint guarantee on each LLM call are unaffected.

3 — The after picture

WfPipeRouter  (parent workflow)
    │
    ├─ pipe_llm._live_run_operator_pipe(...)
    │     └─ get_content_generator()           ◄── returns ContentGeneratorInWorkflow
    │     └─ content_generator.make_llm_text(...)
    │            │
    │            ▼
    │     ContentGeneratorInWorkflow.make_llm_text(...)
    │            │
    │            └─ workflow.execute_activity(             ◄── one frame, not two
    │                   act_llm_gen_text,
    │                   llm_assignment,
    │                   start_to_close_timeout=...,
    │                   retry_policy=...,
    │                   task_queue=inference_task_queue,   ◄── LLM-text only
    │                   activity_id=wfid,                  ◄── for Temporal UI
    │               )
    │
    └───── result

The activities themselves (act_llm_gen_text, act_llm_gen_object, …) are unchanged. The worker registration is unchanged (the seven act_* functions still live in the crafting TaskPack). The only thing that disappears is the layer between the router and the activity.

4 — Why this is a good trade

5 — Key design decisions

def _inference_dispatch_kwargs(worker_config) -> dict[str, Any]:
    """Stopgap. Single deletion point when per-activity routing lands.
    See wip/temporal-primitives/per-activity-queue-routing-v1.md."""
    return {"task_queue": worker_config.inference_task_queue}

6 — Code: the LLM-text dispatch, before vs after

The whole refactor boils down to substituting one Temporal SDK call for another. Below is the LLM-text path. The other six methods follow the identical pattern.

# content_generator_child.py
async def make_llm_text(
    self,
    llm_assignment: LLMAssignment,
    wfid: str = "craft-text",
    ...
) -> str:
    child_workflow_id = make_child_workflow_id(
        parent_id=workflow.info().workflow_id,
        wfid=wfid,
    )
    return await self._workflow_executor \
        .execute_child_workflow(
            WfMakeLLMText,
            arg=llm_assignment,
            id=child_workflow_id,
            ...
        )

# wf_make_llm_text.py
@workflow.defn(name="wf_make_llm_text")
class WfMakeLLMText(
    WorkflowClass[LLMAssignment, str]
):
    @workflow.run
    async def run(
        self, arg: LLMAssignment
    ) -> str:
        wc = get_config().temporal.worker_config
        try:
            return await workflow.start_activity(
                activity=act_llm_gen_text,
                arg=arg,
                start_to_close_timeout=
                    wc.workflow_execution_timeout,
                retry_policy=wc.retry_policy,
                task_queue=wc.inference_task_queue,
            )
        except ActivityError as exc:
            if isinstance(exc.cause, ApplicationError):
                raise TemporalError.from_app_error(
                    exc=exc.cause
                ) from exc
            raise

# content_generator_in_workflow.py
async def make_llm_text(
    self,
    llm_assignment: LLMAssignment,
    wfid: str = "craft-text",
    ...
) -> str:
    self._check_activity_id_unique(wfid)
    wc = get_config().temporal.worker_config
    try:
        # LLM-text only: route to inference queue.
        # See _inference_dispatch_kwargs docstring.
        return await workflow.execute_activity(
            act_llm_gen_text,
            llm_assignment,
            start_to_close_timeout=
                wc.workflow_execution_timeout,
            retry_policy=wc.retry_policy,
            activity_id=wfid,
            **_inference_dispatch_kwargs(wc),
        )
    except ActivityError as exc:
        if isinstance(exc.cause, ApplicationError):
            raise TemporalError.from_app_error(
                exc=exc.cause
            ) from exc
        raise

# wf_make_llm_text.py
#  ──────── DELETED ────────

Notice what stayed identical: the try/except ActivityError → TemporalError.from_app_error(…) translation, the start_to_close_timeout + retry_policy values, and the asymmetric task_queue routing. The behavioral surface is preserved; only the indirection is gone.

7 — The runtime uniqueness check, in code

class ContentGeneratorInWorkflow:
    def __init__(self):
        # Keyed by workflow_id because this generator instance is set once on
        # the hub and reused across many workflow runs. A flat set[str] would
        # produce cross-run false positives.
        self._seen_activity_ids: dict[str, set[str]] = {}
 
    def _check_activity_id_unique(self, activity_id: str) -> None:
        # During replay, Temporal re-walks the history. The set is already
        # populated from the original execution; without this guard, a
        # cache-eviction replay would raise a spurious duplicate.
        if workflow.unsafe.is_replaying():
            return
        wf_id = workflow.info().workflow_id
        seen = self._seen_activity_ids.setdefault(wf_id, set())
        if activity_id in seen:
            msg = f"Duplicate activity_id {activity_id!r} in workflow {wf_id!r}"
            raise ContentGenerationError(msg)
        seen.add(activity_id)

8 — The make_extract_pages page-views fix

This is the one place where the refactor isn't pure deletion — it fixes a pre-existing behavioral divergence between the direct-mode and Temporal-mode generators.

The direct-mode ContentGenerator.make_extract_pages augments its return when extract_job_params.should_include_page_views is true: it calls make_render_page_views(…) for multi-page document_uri inputs, or constructs a single-element page view inline for image_uri inputs.

The Temporal-mode ContentGeneratorChild.make_extract_pages did NOT. It just dispatched WfMakeExtract and returned the page contents as-is, missing both augmentations. Per the project's "flag and fix existing bugs" rule, the collapse is the right moment to fix it.

The new make_extract_pages mirrors the direct generator exactly — and it is the only protocol method that dispatches more than one activity, so it earns special handling for activity_id uniqueness:

async def make_extract_pages(self, ...) -> ExtractPagesOutput:
    base_id = wfid  # e.g. "extract"
    pages_id = f"{base_id}-pages"
    page_views_id = f"{base_id}-render-page-views"
    self._check_activity_id_unique(pages_id)
 
    # First activity: real extraction.
    output = await workflow.execute_activity(
        act_extract_gen_extract_pages, ...,
        activity_id=pages_id,
    )
 
    if not extract_job_params.should_include_page_views:
        return output  # no double-emit when false
 
+    if extract_input.document_uri is not None:
+        # Second activity: render page views from the document.
+        self._check_activity_id_unique(page_views_id)
+        page_views = await workflow.execute_activity(
+            act_render_page_views, ...,
+            activity_id=page_views_id,
+        )
+    elif extract_input.image_uri is not None:
+        # Single image → single-element page view, no second activity.
+        page_views = [ImageContent(url=extract_input.image_uri)]
+    else:
+        page_views = []
+
+    if len(page_views) != len(output.page_contents):
+        raise ContentGenerationError("page_views length mismatch")
+
+    for page_content in output.page_contents:
+        page_content.page_view = page_views.pop(0)
    return output

The two-activity branch is the most fragile post-collapse path — it's the only place where the activity_id uniqueness mitigation actually matters in production. It's covered three ways:

• Unit tests (Phase 1): all three branches — no-page-views, image_uri, document_uri.
• Real-PDF integration test (Phase 4): TestTprlContentGeneratorPdfPageViews exercises the multi-page document_uri branch end-to-end through real Temporal, against a local 2-page PDF (multi-page catches pop(0) ordering bugs that a 1-page PDF would not).
• Cross-process regression guard (Phase 7): TestSplitWorkerExtractPages asserts via WorkflowHandle.fetch_history() that two ActivityTaskScheduled events appear with activity_ids "extract-pages" and "extract-render-page-views" — pins the contract.

9 — File-level inventory

10 — Phase-by-phase landing plan

Each phase ends with make agent-check && make agent-test. Two checkpoints — natural commit boundaries — are placed at A (ready to flip the flag) and B (codebase in target state).

11 — Where bugs are most likely to hide

12 — Follow-up TODOs (out of scope)

• Per-activity, per-handle Temporal queue routing. Replace the provisional inference_task_queue binary model with a general routing system covering LLM, image-gen, and extract — keyed by (activity_name, runtime_handle). Full design in wip/temporal-primitives/per-activity-queue-routing-v1.md. The Phase 7 stopgap helper _inference_dispatch_kwargs is the single deletion point when this lands.
• Watch task-queue saturation post-deploy. Today a PipeSequence of N inference calls schedules N child workflows, decoupling activity scheduling across child task lists. Post-collapse, all N activities run through the parent's task list (or inference_task_queue for LLM-text). For deep pipelines (N > 50), single-queue QPS could spike. Watch ScheduleToStartLatency on both queues. Not blocking; upside dominates.
• Drop make_render_page_views from ContentGeneratorProtocol if no operator caller emerges within one release cycle. Today it's only invoked internally from make_extract_pages; the Protocol method is dead code from the operator side. Estimated savings: ~50 lines across four files.
• Fix make_extract_pages Protocol-violating signature in the direct generator (defaults extract_job_params / extract_job_config to None while the Protocol requires non-None).
• Bound _seen_activity_ids growth. The dict never evicts. A long-running worker accumulates one entry per processed workflow_id over its lifetime. Add an eviction hook on workflow completion, or move the set into workflow-local state via a contextvar set at the start of WfPipeRouter.run.

13 — TL;DR for the reviewer

Companion docs: collapse-content-generation-workflow-layer-v2.md (the why), TODOS.md (the executable plan), per-activity-queue-routing-v1.md (the deletion target for the provisional queue rule).