# Execute Grox content plans

> Engine, Dispatcher, and GrpcServer startup, PlanMaster fan-out, TaskEligibility gating, and merged TaskResult fields.

- Repository: xai-org/x-algorithm
- GitHub: https://github.com/xai-org/x-algorithm
- Human docs: https://grok-wiki.com/public/docs/xai-org-x-algorithm-23c09c39074c
- Complete Markdown: https://grok-wiki.com/public/docs/xai-org-x-algorithm-23c09c39074c/llms-full.txt

## Source Files

- `grox/main.py`
- `grox/engine.py`
- `grox/dispatcher.py`
- `grox/plans/plan.py`
- `grox/plans/plan_master.py`
- `grox/schedules/types.py`
- `grox/plans/plan_post_safety.py`

---

---
title: "Execute Grox content plans"
description: "Engine, Dispatcher, and GrpcServer startup, PlanMaster fan-out, TaskEligibility gating, and merged TaskResult fields."
---

`grox/main.py` `serve()` builds one shared `ScheduleContext`, then starts `Engine`, `Dispatcher`, and `GrpcServer` in that order. The dispatcher process pulls Kafka-backed `TaskPayload`s, injects `TaskEligibility` values, and writes them to `task_queue`. The engine process reads that queue, runs `PlanMaster.exec`, and writes a merged `TaskResult` to `resp_queue`. The published snapshot does not include `grox.service`, `grox.config`, `grox.data_loaders.data_types`, `grox.data_loaders.media_processor`, or `monitor`, so `python grox/main.py` cannot start from this checkout.

<Warning>
This is a production-shaped snapshot, not a local runner. Phoenix inference in `phoenix/run_pipeline.py` is the checkout that executes. Grox plan execution needs the unpublished config, gRPC, media, and metrics modules listed under [Runtime boundary](#runtime-boundary).
</Warning>

## Process layout

Three OS processes share a `multiprocessing.Manager` dict. Child processes call `init_proc`, which ignores `SIGINT` / `SIGTERM`; only the main `serve()` loop handles those signals.

```mermaid
flowchart TB
  subgraph Main["main process — grox/main.py serve()"]
    Init["init_proc('main')"]
    Ctx["new_context()"]
    Signals["SIGINT / SIGTERM → shutdown Event"]
    Grpc["GrpcServer(context)"]
  end

  subgraph Shared["ScheduleContext manager.dict"]
    TQ["task_queue"]
    RQ["resp_queue"]
    SE["shutdown_event"]
    QSE["queue_connection_shutdown_event"]
    LTQ["live_task_queue"]
    LRQ["live_resp_queue"]
  end

  subgraph Disp["grox-dispatcher process"]
    Fill["_fill_loop"]
    Result["_result_loop"]
    Gens["PriorityTaskGenerator"]
    Loaders["KafkaPostLoader / MessageQueueLoader"]
  end

  subgraph Eng["grox-engine process"]
    Poll["_poll_task"]
    PM["PlanMaster.exec"]
    Plans["ALL_PLANS DAGs"]
    ASR["ASRProcessor.start"]
    Media["MediaProcessor.start"]
  end

  Kafka["Kafka topics via KafkaTopicName"] --> Loaders
  Loaders --> Gens
  Gens --> Fill
  Fill --> TQ
  TQ --> Poll
  Poll --> PM
  PM --> Plans
  Plans --> RQ
  RQ --> Result
  Result --> Gens
  Init --> Ctx
  Ctx --> Shared
  Signals --> QSE
  QSE --> Gens
  Signals --> SE
  SE --> Poll
  SE --> Fill
  Grpc --> Shared
```

`Engine` and `Dispatcher` only use `task_queue`, `resp_queue`, `shutdown_event`, and `queue_connection_shutdown_event`. `live_task_queue` and `live_resp_queue` are allocated and unused by those two classes. `GrpcServer` is constructed with the same context; its RPCs are not in this tree.

## Start the service

<Steps>
<Step title="Entry point">
The published entry is `grox/main.py`:

```python
if __name__ == "__main__":
    asyncio.run(serve())
```

There is no `grox/` `pyproject.toml` or console script in this checkout.
</Step>
<Step title="Shared context">
`new_context()` returns a `SyncManager` dict with six keys:

| Key | Type | Used by |
|---|---|---|
| `task_queue` | `Queue[TaskPayload]` | Dispatcher put, Engine get |
| `resp_queue` | `Queue[TaskResult]` | Engine put, Dispatcher get |
| `live_task_queue` | `Queue` | Allocated only |
| `live_resp_queue` | `Queue` | Allocated only |
| `shutdown_event` | `Event` | Engine / Dispatcher loops |
| `queue_connection_shutdown_event` | `Event` | Dispatcher generator stop |
</Step>
<Step title="Start order">
`serve()` waits for each child `started_event` before continuing:

1. `Engine.start()` — process name `grox-engine`. Child runs `init_proc("engine")`, then `MediaProcessor.start()` and `ASRProcessor.start()`, then sets `started_event`.
2. `Dispatcher.start()` — process name `grox-dispatcher`. Child `init_proc("dispatcher")` is wrapped in tenacity: 3 attempts, wait 1s then +3s (cap 9s). It builds `PriorityTaskGenerator` from `grox_config.dispatcher.task_generators` and calls `start()` on every generator (each starts its Kafka loader).
3. `GrpcServer(context).start()` — implementation is unpublished.

Logs `Grox server started` only after all three `start()` calls return.
</Step>
<Step title="Verify">
Expected log sequence on a complete deployment:

```text
Starting grox server...
Starting Grox engine...
Grox engine started
Starting Grox dispatcher...
Grox dispatcher started
Grox server started
```

An `ImportError` for `grox.service`, `grox.config.config`, `monitor.logging`, or `grox.data_loaders.media_processor` is the local-checkout failure mode.
</Step>
</Steps>

## Dispatcher intake

`Dispatcher` runs three coroutines: `_fill_loop`, `_result_loop`, and `_wait_for_queue_connection_shutdown`.

### Generator selection

`grox_config.dispatcher.task_generators` is a list of `{type, max_qps, weight}`. Unknown `TaskGeneratorType` raises `ValueError`. `PriorityTaskGenerator` requires a non-empty list and strictly positive weights. Each poll round samples remaining generators with `random.choices` by weight; a generator that yields `None` is dropped from that round only.

`TaskGenerator.poll` applies an in-process `FixedWindowRateLimiter` when `max_qps` is set (`RateLimitItemPerSecond(max_qps, 1)`).

### Eligibility injection

`StreamTaskGenerator._poll` copies loader fields onto `TaskPayload` and sets `eligibilities` from the generator class constant. `payload_id` is a new `uuid.uuid4().hex` from `KafkaPostLoader`, not the Kafka offset.

| `TaskGeneratorType` | Injected `TaskEligibility` |
|---|---|
| `POST_STREAM` | `spam_comment`, `reply_ranking` |
| `POST_STREAM_RECOVERY` | `banger_initial_screen` |
| `POST_STREAM_TEST` | `banger_initial_screen` |
| `POST_STREAM_DELAYED` | none (`{}`) |
| `POST_SAFETY_STREAM` | `post_safety` |
| `POST_MIN_TRACTION_STREAM_FOR_GROX` | `banger_initial_screen` |
| `POST_MIN_TRACTION_STREAM_FOR_GROX_PTOS` | `safety_ptos` |
| `POST_MIN_TRACTION_STREAM_FOR_GROX_MULTI_MODAL` | `post_embedding_with_summary_for_reply` |
| `POST_EMBEDDING_REQUEST_STREAM_WITH_SUMMARY` | `post_embedding_with_summary` |
| `POST_EMBEDDING_REQUEST_STREAM_WITH_SUMMARY_RECOVERY` | `post_embedding_with_summary` |
| `POST_EMBEDDING_REQUEST_STREAM_WITH_SUMMARY_FOR_REPLY_RECOVERY` | `post_embedding_with_summary_for_reply` |
| `POST_EMBEDDING_V5_STREAM` | `mm_emb_v5` |
| `POST_EMBEDDING_V5_FOR_REPLY_STREAM` | `mm_emb_v5_for_reply` |
| `REPLY_RANKING_RECOVERY` | `reply_ranking` |
| `SAFETY_PTOS_RECOVERY` | `safety_ptos` |
| `SAFETY_PTOS_DELUXE` | `safety_ptos` |

`TaskEligibility.MM_EMB_V4` exists on the enum and is not injected by any published generator and not required by any published plan.

<Warning>
`POST_STREAM_DELAYED` injects an empty eligibility set. Every `Plan.execute` returns `None`, `PlanMaster.merge_results` then calls `min` / `max` on an empty list, and the engine records a failed `TaskResult`.
</Warning>

### In-flight cap and retry

<ParamField body="max_in_flight" type="int" required>
`grox_config.dispatcher.max_in_flight`. Fill loop sleeps 10ms while `len(_in_flights) >= max_in_flight`.
</ParamField>

<ParamField body="max_attempts" type="int" required>
`grox_config.dispatcher.max_attempts`. Compared against `TaskPayload.attempt` (starts at `0`).
</ParamField>

<ParamField body="graceful_shutdown_timeout" type="float" required>
`grox_config.dispatcher.graceful_shutdown_timeout`. Parent `Process.join` timeout.
</ParamField>

```mermaid
sequenceDiagram
  participant Kafka
  participant Gen as StreamTaskGenerator
  participant Fill as Dispatcher._fill_loop
  participant TQ as task_queue
  participant Eng as Engine
  participant PM as PlanMaster
  participant RQ as resp_queue
  participant Res as Dispatcher._result_loop

  Kafka->>Gen: MessageQueuePayload
  Gen->>Gen: eligibilities = ELIGIBILITIES_TO_INJECT
  Fill->>Fill: wait while inflight >= max_in_flight
  Fill->>TQ: put TaskPayload
  Eng->>TQ: get_nowait
  Eng->>PM: exec(task)
  PM->>PM: gather ALL_PLANS.execute
  PM->>PM: merge_results(non-None)
  Eng->>RQ: put TaskResult
  Res->>RQ: get_nowait
  alt success
    Res->>Gen: ack(payload_id, success=True)
  else attempt < max_attempts
    Res->>TQ: put same payload, attempt += 1
  else final failure
    Res->>Gen: ack(payload_id, success=False) if origin known
  end
```

On success the payload id is removed from `_in_flights` and `PriorityTaskGenerator.ack` pops the origin label. On retry the id stays in `_in_flights` and is put on `task_queue` again. On final failure the dispatcher looks up origin via `_result_cache`; if missing, it logs and skips `ack`.

Published `KafkaLoader.ack` is a no-op (`pass`). The dispatcher still calls it.

## Engine execution

`Engine._run` non-blocking-gets from `task_queue`. Empty queue sleeps 100ms. Each payload is `asyncio.create_task`'d with no engine-side concurrency cap — the dispatcher in-flight set is the only backpressure.

`_process_task` is `PlanMaster.exec(task)`. Logging context keys: `task=payload_id`, `post=post.id` when present, `user=user.id` or `user_context.user.id`. A tracer span `task.root` is opened under `Metrics.tracer("engine")`.

<ResponseField name="engine exception TaskResult" type="TaskResult">
If `PlanMaster.exec` raises, the engine still puts a `TaskResult` so the dispatcher can retry: `success=False`, `error=str(e)`, `task_finished_at` set to the pre-task `perf_counter`, `task_started_at` set to now. Those two timestamps are swapped relative to the field names.
</ResponseField>

Loop condition: `while not shutdown_event or not task_queue.empty()`. After the loop, `run()` calls `os._exit(0)` without awaiting outstanding `_run_task` tasks. In-flight plan work can be killed once the queue is empty and shutdown is set.

`Engine.stop()` in the parent joins `grox-engine` for `grox_config.engine.graceful_shutdown_timeout`, then awaits `MediaProcessor.stop()` and `ASRProcessor.stop()` (ASR stop default timeout is 5s). Those processors were started inside the child.

## PlanMaster fan-out

`PlanMaster.ALL_PLANS` is a class-level list of already-constructed plan instances, in this order:

1. `PlanInitialBanger`
2. `PlanPostSafety`
3. `PlanSpamComment`
4. `PlanPostEmbeddingWithSummary`
5. `PlanPostEmbeddingWithSummaryForReply`
6. `PlanPostEmbeddingV5`
7. `PlanPostEmbeddingV5ForReply`
8. `PlanReplyRanking`
9. `PlanSafetyPtos`

`exec` gathers every `plan.execute(task)` concurrently on the same `TaskPayload`. Ineligible plans return `None` and are dropped before merge. Gather order matches `ALL_PLANS`, which is also merge order.

## TaskEligibility gating

Each `Plan` declares `REQUIRED_ELIGIBILITY`. `Plan.execute` returns `None` immediately when that value is not in `task.eligibilities`. No `plan.execute.*` metrics are recorded for a skipped plan.

| Plan | `REQUIRED_ELIGIBILITY` |
|---|---|
| `PlanInitialBanger` | `banger_initial_screen` |
| `PlanPostSafety` | `post_safety` |
| `PlanSpamComment` | `spam_comment` |
| `PlanPostEmbeddingWithSummary` | `post_embedding_with_summary` |
| `PlanPostEmbeddingWithSummaryForReply` | `post_embedding_with_summary_for_reply` |
| `PlanPostEmbeddingV5` | `mm_emb_v5` |
| `PlanPostEmbeddingV5ForReply` | `mm_emb_v5_for_reply` |
| `PlanReplyRanking` | `reply_ranking` |
| `PlanSafetyPtos` | `safety_ptos` |

One payload can carry multiple eligibilities. `POST_STREAM` therefore runs `PlanSpamComment` and `PlanReplyRanking` in the same `PlanMaster.exec` call; the other seven plans return `None`.

Plan and task DAGs, classifier wiring, and per-plan `TASK_DEPENDENCIES` live on [Grox plans and tasks](/grox-plans-and-tasks). The runtime rule that matters here: every task name in `TASKS` is launched concurrently; a task waits on dependency futures; a dependency result of `TaskResultCategory.SKIPPED` skips the dependent; a raised exception is set on the future, re-raised, appended to `TaskContext.errors`, and marks the plan `success=False`. Leaf tasks (nothing depends on them) still run. `Task.exec` itself retries twice with a 1s wait via tenacity.

## Merged TaskResult

<ParamField body="payload_id" type="string" required>
Stable id for inflight tracking and ack. New hex UUID per Kafka message.
</ParamField>

<ParamField body="eligibilities" type="set[TaskEligibility]">
Copied from the stream generator. Default empty set.
</ParamField>

<ParamField body="attempt" type="int">
Retry counter. Starts at `0`. Dispatcher increments before resubmit.
</ParamField>

<ParamField body="task_type" type="TaskGeneratorType | None">
Generator type that created the payload. Used by some deluxe/PTOS tasks.
</ParamField>

<ParamField body="deadline_ts_secs" type="int | None">
`now + loader_config.task_deadline_secs` at Kafka decode. Carried on `TaskPayload`; Engine and Plan do not enforce it.
</ParamField>

Eligible `Plan.execute` always returns a `TaskResult`:

<ResponseField name="task" type="TaskPayload">
The original payload.
</ResponseField>

<ResponseField name="task_started_at" type="float">
`TaskContext.start_time` (`perf_counter` at context create). Merge takes `min`.
</ResponseField>

<ResponseField name="task_finished_at" type="float">
`perf_counter` at plan return. Merge takes `max`.
</ResponseField>

<ResponseField name="content_categories" type="list[ContentCategoryResult]">
Merge concatenates every plan list, each entry `model_copy()`.
</ResponseField>

<ResponseField name="multimodal_post_embedding" type="list[float] | None">
Merge keeps the first non-`None` embedding in `ALL_PLANS` order. Later embeddings are discarded.
</ResponseField>

<ResponseField name="reason" type="string">
Merge joins non-empty plan reasons with `\n`.
</ResponseField>

<ResponseField name="success" type="bool">
Plan: `len(ctx.errors) == 0`. Merge: `all(r.success)`.
</ResponseField>

<ResponseField name="error" type="string | None">
Plan: `"\n".join(str(e) for e in ctx.errors)`. Merge: joins `r.error or "unknown error"` for unsuccessful plans only.
</ResponseField>

<RequestExample>
```json
{
  "payload_id": "a1b2c3d4e5f64789a0b1c2d3e4f50617",
  "attempt": 0,
  "task_type": "POST_STREAM",
  "eligibilities": ["spam_comment", "reply_ranking"],
  "deadline_ts_secs": 1773600000
}
```
</RequestExample>

<ResponseExample>
```json
{
  "success": true,
  "error": "",
  "reason": "",
  "content_categories": [],
  "multimodal_post_embedding": null,
  "task_started_at": 1024.11,
  "task_finished_at": 1026.40
}
```
</ResponseExample>

<Warning>
`merge_results` assumes `results` is non-empty. Zero eligible plans (`POST_STREAM_DELAYED`, or a payload whose eligibilities match no plan) raises inside `min` / `max` and becomes an engine-level failed `TaskResult`.
</Warning>

## Shutdown

<Steps>
<Step title="Signal">
Main `serve()` registers `SIGINT` and `SIGTERM` to set a local `asyncio.Event` named `shutdown`. Child processes ignore those signals.
</Step>
<Step title="Stop intake, then wait 300s">
After the signal, `queue_connection_shutdown_context` sets `queue_connection_shutdown_event`. Dispatcher stops all task generators (Kafka loaders). Main then `await asyncio.sleep(300)` before touching `shutdown_event`. Engine keeps consuming `task_queue` during that window.
</Step>
<Step title="Stop workers">
`shutdown_context` sets `shutdown_event`. Dispatcher fill loop exits; result loop continues while `_in_flights` is non-empty. Engine loop exits when the queue is empty, then `os._exit(0)`.
</Step>
<Step title="Join">
Main gathers `grpc_server.stop()`, `dispatcher.stop()`, `engine.stop()`, then `cleanup()` shuts down the `SyncManager`. Join timeouts come from `grox_config.*.graceful_shutdown_timeout`.
</Step>
</Steps>

Expected teardown logs: `Grox server shutting down...` then `Grox server stopped`.

## Referenced config

`grox.config.config.grox_config` is unpublished. These attributes are read by the published execution path:

| Attribute | Reader |
|---|---|
| `dispatcher.task_generators[].type` | `Dispatcher._get_task_generators` |
| `dispatcher.task_generators[].max_qps` | generator constructor |
| `dispatcher.task_generators[].weight` | `PriorityTaskGenerator` |
| `dispatcher.max_in_flight` | fill loop |
| `dispatcher.max_attempts` | result loop |
| `dispatcher.graceful_shutdown_timeout` | `Dispatcher.stop` |
| `engine.graceful_shutdown_timeout` | `Engine.stop` |
| `logging` | `init_proc` → `Logging.config` |
| `metrics` | `init_proc` → `Metrics.init` |
| `periodic_gc.interval`, `periodic_gc.jitter` | `periodic_gc` task |
| `asr.max_workers` | `ASRProcessor.start` |
| `get_kafka_loader_topic` / `get_kafka_consumer_topic` | `KafkaLoader` |
| loader `prefetching_threshold`, `prefetching_batch_size`, `task_deadline_secs` | Kafka prefetcher / deadline |

## Metrics

| Name | Kind | When |
|---|---|---|
| `dispatcher.inflight.count` | gauge | after add/remove of `_in_flights` |
| `dispatcher.task.sent.count` | counter | every put, attribute `task_type` |
| `dispatcher.result.received.count` | counter | every `resp_queue` get |
| `dispatcher.result.success.count` | counter | `result.success` |
| `dispatcher.result.failed.count` | counter | retry path |
| `dispatcher.result.failed.final.count` | counter | exhausted retries, attribute `origin` |
| `engine.task.received.count` | counter | successful `task_queue` get |
| `engine.task.success.count` | counter | `PlanMaster.exec` returned |
| `engine.task.failed.count` | counter | exception wrapper |
| `engine.task.processing_time` | histogram | seconds around `PlanMaster.exec` |
| `plan.execute.count` | counter | eligible plan start, `plan_name` |
| `plan.execute.success.count` | counter | no exception in gather |
| `plan.execute.failed.count` | counter | gather exception |
| `plan.execute.duration` | histogram | eligible plan wall time |
| `task.exec.count` / `.intaken` / `.success` / `.skipped` / `.failed` | counter | per `Task.exec`, `task_name` |

`plan_name` is `camel_to_snake` of the plan class (`PlanPostSafety` → `plan_post_safety`).

## Runtime boundary

These imports are required to start or execute plans and are not present under `grox/`:

- `grox.service.GrpcServer`
- `grox.config.config` (`grox_config`, `TaskGeneratorType`, `KafkaTopicName`)
- `grox.config.env` (task disable rules)
- `grox.data_loaders.data_types`
- `grox.data_loaders.media_processor.MediaProcessor`
- `monitor.logging`, `monitor.metrics`
- `kafka_cli.consumer`, `kafka_cli.config`, `thrifts.serdes`

`BrokenPipeError` on a manager `Event` or queue is treated as shutdown (`Engine` / `Dispatcher` `_is_shutdown` return `True`; poll methods return `None`).

## Failure modes

| Symptom | Cause | What to check |
|---|---|---|
| `ImportError` on `grox.service` / `grox.config` / `monitor` | Unpublished modules | This checkout cannot run Grox |
| `ValueError: Invalid task generator type` | Unknown `TaskGeneratorType` in config | Dispatcher match arms listed above |
| `ValueError: No generators provided` | Empty `task_generators` | Config list must be non-empty |
| `ValueError: All weights must be positive` | `weight <= 0` | Priority mixer |
| `ValueError: Not every task in TASK_DEPENDENCIES is defined in TASKS` | Plan DAG / `TASKS` mismatch at construct | Plan class body |
| Engine failed result with empty plans | No matching eligibility (`POST_STREAM_DELAYED` or unknown set) | `merge_results` empty `min`/`max` |
| Fill loop idle, engine idle | `len(_in_flights) >= max_in_flight` | `dispatcher.inflight.count` |
| Task retried then dropped | `attempt` reached `max_attempts` | `dispatcher.result.failed.final.count` |
| `No origin found ... skipping ack` | Missing `_result_cache` label | Priority mixer cache vs payload id |
| Child ignores Ctrl-C | `prevent_default()` in `init_proc` | Signal the main process |
| Engine dies mid-plan after drain | `os._exit(0)` when queue empty | 300s window then shutdown_event |
| Kafka ack appears to do nothing | `KafkaLoader.ack` is `pass` | Offset commit is unpublished |

## Next

<CardGroup>
<Card title="Grox plans and tasks" href="/grox-plans-and-tasks">
`ALL_PLANS`, per-plan DAGs, `TaskEligibility` values, classifiers, and `TaskResult` merge rules.
</Card>
<Card title="Runtime boundaries" href="/runtime-boundaries">
What this checkout can run locally versus Home Mixer, Thunder, and Grox snapshots that need unpublished modules.
</Card>
<Card title="Troubleshooting" href="/troubleshooting">
LFS/artifact failures, empty feeds, Thunder `RESOURCE_EXHAUSTED`, and unpublished Grox imports.
</Card>
</CardGroup>