# Grox plans and tasks

> PlanMaster.ALL_PLANS, TaskEligibility values, DAG TASK_DEPENDENCIES, ContentClassifier, and TaskResult merge rules.

- 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/plans/plan_master.py`
- `grox/plans/plan.py`
- `grox/schedules/types.py`
- `grox/dispatcher.py`
- `grox/engine.py`
- `grox/tasks/task.py`
- `grox/classifiers/content/classifier.py`
- `grox/plans/plan_post_safety.py`

---

---
title: "Grox plans and tasks"
description: "PlanMaster.ALL_PLANS, TaskEligibility values, DAG TASK_DEPENDENCIES, ContentClassifier, and TaskResult merge rules."
---

`PlanMaster.exec` fans every inbound `TaskPayload` across `ALL_PLANS` in parallel. Each `Plan` runs only when its `REQUIRED_ELIGIBILITY` is in `payload.eligibilities`, then executes a named `TASKS` / `TASK_DEPENDENCIES` DAG. Non-`None` plan `TaskResult`s are folded by `PlanMaster.merge_results` before `Engine` puts the merged result on `resp_queue`.

<Info>
Startup, process layout, and Kafka generator wiring live on [Execute Grox content plans](/execute-grox-plans). This page is the plan/task contract: eligibility values, DAGs, classifiers, and merge rules.
</Info>

<Warning>
`grox/main.py` imports unpublished modules (`grox.service.GrpcServer`, `grox.config.config`, `grox.data_loaders.data_types`, `grok_sampler`, `monitor`, `strato_http`). The plan graph is readable in this checkout; the service does not run from the published tree alone.
</Warning>

## Architecture

```mermaid
flowchart TB
  subgraph generators [Stream generators]
    SG["StreamTaskGenerator.ELIGIBILITIES_TO_INJECT"]
  end
  subgraph runtime [Dispatcher and Engine]
    TQ["task_queue: TaskPayload"]
    EN["Engine._process_task"]
    RQ["resp_queue: TaskResult"]
  end
  subgraph master [PlanMaster]
    ALL["asyncio.gather ALL_PLANS"]
    MR["merge_results"]
  end
  subgraph planLayer [Plan]
    EL["_eligible(REQUIRED_ELIGIBILITY)"]
    DAG["TASK_DEPENDENCIES futures"]
  end
  subgraph taskLayer [Task]
    SK["should_skip / DISABLE_RULES"]
    EX["_exec → TaskContext"]
  end
  SG --> TQ --> EN --> ALL
  ALL --> EL
  EL -->|ineligible: None| MR
  EL -->|eligible| DAG --> SK --> EX
  EX --> MR --> RQ
```

`Dispatcher` writes `TaskPayload`s. `Engine._process_task` is the only caller of `PlanMaster.exec`. `Plan.execute` always returns a `TaskResult` when eligible; ineligible plans return `None` and are dropped before merge.

## TaskEligibility

`TaskEligibility` is a `str` enum on `TaskPayload.eligibilities`. A plan runs iff `REQUIRED_ELIGIBILITY in payload.eligibilities` — membership, not equality. One payload can activate multiple plans.

| Member | Value | Plan |
|---|---|---|
| `SPAM_COMMENT` | `spam_comment` | `PlanSpamComment` |
| `BANGER_INITIAL_SCREEN` | `banger_initial_screen` | `PlanInitialBanger` |
| `POST_EMBEDDING_WITH_SUMMARY` | `post_embedding_with_summary` | `PlanPostEmbeddingWithSummary` |
| `POST_EMBEDDING_WITH_SUMMARY_FOR_REPLY` | `post_embedding_with_summary_for_reply` | `PlanPostEmbeddingWithSummaryForReply` |
| `MM_EMB_V5` | `mm_emb_v5` | `PlanPostEmbeddingV5` |
| `MM_EMB_V5_FOR_REPLY` | `mm_emb_v5_for_reply` | `PlanPostEmbeddingV5ForReply` |
| `REPLY_RANKING` | `reply_ranking` | `PlanReplyRanking` |
| `SAFETY_PTOS` | `safety_ptos` | `PlanSafetyPtos` |
| `POST_SAFETY` | `post_safety` | `PlanPostSafety` |
| `MM_EMB_V4` | `mm_emb_v4` | **No plan.** `TaskMultimodalPostEmbeddingRecsysV4` exists but is not registered in any `TASKS` map. |

### Who injects which eligibility

`StreamTaskGenerator._poll` copies `ELIGIBILITIES_TO_INJECT` onto every payload.

| Generator | `TaskGeneratorType` | Injected eligibilities |
|---|---|---|
| `PostStreamTaskGenerator` | `POST_STREAM` | `SPAM_COMMENT` **and** `REPLY_RANKING` |
| `MinTractionPostStreamForGroxTaskGenerator` | `POST_MIN_TRACTION_STREAM_FOR_GROX` | `BANGER_INITIAL_SCREEN` |
| `PostStreamRecoveryTaskGenerator` | `POST_STREAM_RECOVERY` | `BANGER_INITIAL_SCREEN` |
| `PostStreamTestTaskGenerator` | `POST_STREAM_TEST` | `BANGER_INITIAL_SCREEN` |
| `PostSafetyStreamTaskGenerator` | `POST_SAFETY_STREAM` | `POST_SAFETY` |
| `PlanSpamComment` / reply-ranking recovery | `REPLY_RANKING_RECOVERY` | `REPLY_RANKING` |
| `PostEmbeddingRequestWithSummaryStreamTaskGenerator` | `POST_EMBEDDING_REQUEST_STREAM_WITH_SUMMARY` | `POST_EMBEDDING_WITH_SUMMARY` |
| `PostEmbeddingRequestWithSummaryRecoveryStreamTaskGenerator` | `POST_EMBEDDING_REQUEST_STREAM_WITH_SUMMARY_RECOVERY` | `POST_EMBEDDING_WITH_SUMMARY` |
| `MinTractionPostStreamForGroxMultiModalTaskGenerator` | `POST_MIN_TRACTION_STREAM_FOR_GROX_MULTI_MODAL` | `POST_EMBEDDING_WITH_SUMMARY_FOR_REPLY` |
| `PostEmbeddingRequestWithSummaryForReplyRecoveryStreamTaskGenerator` | `POST_EMBEDDING_REQUEST_STREAM_WITH_SUMMARY_FOR_REPLY_RECOVERY` | `POST_EMBEDDING_WITH_SUMMARY_FOR_REPLY` |
| `PostEmbeddingV5StreamTaskGenerator` | `POST_EMBEDDING_V5_STREAM` | `MM_EMB_V5` |
| `PostEmbeddingV5ForReplyStreamTaskGenerator` | `POST_EMBEDDING_V5_FOR_REPLY_STREAM` | `MM_EMB_V5_FOR_REPLY` |
| `MinTractionPostStreamForGroxPtosTaskGenerator` | `POST_MIN_TRACTION_STREAM_FOR_GROX_PTOS` | `SAFETY_PTOS` |
| `SafetyPtosRecoveryStreamTaskGenerator` | `SAFETY_PTOS_RECOVERY` | `SAFETY_PTOS` |
| `SafetyPtosDeluxeStreamTaskGenerator` | `SAFETY_PTOS_DELUXE` | `SAFETY_PTOS` |
| `PostStreamDelayedTaskGenerator` | `POST_STREAM_DELAYED` | `{}` — no plan is eligible |

<Note>
`POST_STREAM` is the only generator that injects two eligibilities. `PlanMaster` then runs `PlanSpamComment` and `PlanReplyRanking` concurrently on the same payload. Deluxe PTOS reuses `SAFETY_PTOS`; deluxe vs standard is selected later by `payload.task_type == TaskGeneratorType.SAFETY_PTOS_DELUXE`.
</Note>

## PlanMaster.ALL_PLANS

Registration order in `grox/plans/plan_master.py` is also merge order for the first non-`None` embedding:

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

```python
results = await asyncio.gather(*[p.execute(task) for p in cls.ALL_PLANS])
result = cls.merge_results(task, [r for r in results if r is not None])
```

`Plan.get_name()` is `camel_to_snake(class_name)` and is the `plan_name` metric attribute (`plan_initial_banger`, `plan_post_safety`, …). Construction fails if any `TASK_DEPENDENCIES` key or edge is missing from `TASKS`.

## DAG execution

`Plan.execute` creates one `asyncio.Future` per task that appears as a dependency (`self.deps`). Every `TASKS` key is scheduled concurrently; a task `await`s its upstream futures, then calls `task.exec(ctx)`.

| Upstream result | Downstream |
|---|---|
| Any dep returns `TaskResultCategory.SKIPPED` | Downstream is marked `SKIPPED` and does **not** run |
| Any dep raises | Future is set with the exception; the plan `gather` fails |
| All deps `SUCCESS` (or no deps) | `task.exec(ctx)` runs |

Plan-level outcome:

- Exceptions are caught, appended to `ctx.errors`, and do **not** escape `Plan.execute`.
- Remaining incomplete futures are cancelled in `finally`.
- Returned `TaskResult.success` is `len(ctx.errors) == 0`.
- Metrics: `plan.execute.count`, `.success.count`, `.failed.count`, `plan.execute.duration`.

```text
filter ──► rate_limit ──► media ──► classify/embed ──┬──► sink / Manhattan
                                                     └──► Kafka / UPA
```

Most plans follow that spine. A skipped filter or rate-limit (`TaskStopExecution`) short-circuits the rest of the DAG without failing the plan.

## Plan DAG catalog

<AccordionGroup>
<Accordion title="PlanInitialBanger — BANGER_INITIAL_SCREEN">

| Task key | Class |
|---|---|
| `task_initial_banger_filter` | `TaskInitialBangerFilter` |
| `task_banger_annotation_rate_limit` | `TaskRateLimitBangerAnnotationWithPost` |
| `task_media_hydration` | `TaskMediaHydrationBanger` |
| `task_banger_screen_initial` | `TaskBangerScreen` |
| `task_grok_upa_action_with_labels` | `TaskGrokUpaActionWithLabels` |
| `task_publish_unified_post_annotations_manhattan` | `TaskPublishUnifiedPostAnnotationsManhattan` |
| `task_publish_kafka` | `TaskPublishKafka` |

```text
task_initial_banger_filter
  → task_banger_annotation_rate_limit
    → task_media_hydration
      → task_banger_screen_initial
          ├─→ task_grok_upa_action_with_labels
          └─→ task_publish_unified_post_annotations_manhattan
                → task_publish_kafka
```

Filter drops replies and protected authors. Screen writes `ContentCategoryType.BANGER_INITIAL_SCREEN` and `GROK_RANKER` (positive when `quality_score >= 0.4`).

</Accordion>
<Accordion title="PlanPostSafety — POST_SAFETY">

| Task key | Class |
|---|---|
| `task_post_safety_deluxe_filter` | `TaskPostSafetyDeluxeFilter` |
| `task_post_safety_annotation_rate_limit` | `TaskRateLimitPostSafetyAnnotationWithPost` |
| `task_media_hydration` | `TaskMediaHydrationBanger` |
| `task_post_safety_screen_deluxe` | `TaskPostSafetyScreenDeluxe` |
| `task_grok_upa_action_with_labels` | `TaskGrokUpaActionWithLabels` |
| `task_upsert_tweet_bool_metadata_to_unified_post_annotations_manhattan` | `TaskUpsertTweetBoolMetadataToUnifiedPostAnnotation` |

```text
task_post_safety_deluxe_filter
  → task_post_safety_annotation_rate_limit
    → task_media_hydration
      → task_post_safety_screen_deluxe
          ├─→ task_grok_upa_action_with_labels
          └─→ task_upsert_tweet_bool_metadata_to_unified_post_annotations_manhattan
```

Filter drops replies and protected authors. Classifier always emits `POST_SAFETY_SCREEN` with `positive=False`, `score=0.0`; labels live on `tweet_bool_metadata`.

</Accordion>
<Accordion title="PlanSpamComment — SPAM_COMMENT">

| Task key | Class |
|---|---|
| `task_spam_filter` | `TaskSpamFilter` |
| `task_reply_spam_annotation_rate_limit` | `TaskRateLimitReplySpamAnnotationWithPost` |
| `task_media_hydration` | `TaskMediaHydration` |
| `task_spam_detection` | `TaskSpamDetection` |
| `task_publish_reply_spam_mh` | `TaskWriteReplySpamManhattan` |
| `task_publish_kafka` | `TaskPublishKafka` |

```text
task_spam_filter
  → task_reply_spam_annotation_rate_limit
    → task_media_hydration
      → task_spam_detection
          ├─→ task_publish_reply_spam_mh
          └─→ task_publish_kafka
```

Filter requires a reply with a distinct author from both the parent and the root. Detection writes `ContentCategoryType.SPAM_COMMENT`.

</Accordion>
<Accordion title="PlanReplyRanking — REPLY_RANKING">

| Task key | Class |
|---|---|
| `task_reply_ranking_filter` | `TaskReplyRankingFilter` |
| `task_reply_ranking_annotation_rate_limit` | `TaskRateLimitReplyRankingAnnotationWithPost` |
| `task_media_hydration` | `TaskMediaHydration` |
| `task_rank_replies` | `TaskRankReplies` |
| `task_write_reply_ranking_manhattan` | `TaskWriteReplyRankingManhattan` |

```text
task_reply_ranking_filter
  → task_reply_ranking_annotation_rate_limit
    → task_media_hydration
      → task_rank_replies
        → task_write_reply_ranking_manhattan
```

Scores go to `ctx.reply_ranking_results`, not `content_categories`. `ReplyScorer` is not a `ContentClassifier`.

</Accordion>
<Accordion title="PlanSafetyPtos — SAFETY_PTOS">

| Task key | Class |
|---|---|
| `task_safety_ptos_filter` | `TaskSafetyPtosFilter` |
| `task_safety_ptos_annotation_rate_limit` | `TaskRateLimitSafetyPtosAnnotationWithPost` |
| `task_media_hydration` | `TaskMediaHydration` |
| `task_safety_ptos_category_detection` | `TaskSafetyPtosCategoryDetection` |
| `task_safety_ptos_policy_detection` | `TaskSafetyPtosPolicyDetection` |
| `task_write_safety_post_annotations_result_sink` | `TaskWriteSafetyPostAnnotationsResultSink` |

```text
task_safety_ptos_filter
  → task_safety_ptos_annotation_rate_limit
    → task_media_hydration
      → task_safety_ptos_category_detection
        → task_safety_ptos_policy_detection
          → task_write_safety_post_annotations_result_sink
```

Category detection writes `ctx.safety_annotations`. Policy detection fills `violatedPolicies[].safetyPolicy`. Deluxe (`task_type == SAFETY_PTOS_DELUXE`) uses `VLM_PRIMARY_CRITICAL` plus an AdultContent recheck injected at score `50` and removed if the policy comes back `NoViolation`. The filter root uses `{}` (empty dict) rather than `set()`; iteration still yields no deps.

</Accordion>
<Accordion title="Embedding plans — summary v3 and v5">

**`PlanPostEmbeddingWithSummary`** (`POST_EMBEDDING_WITH_SUMMARY`) — originals only:

```text
task_post_embedding_rate_limit_summary
  → task_post_embedding_with_summary_filter
    → task_media_hydration
      → task_post_embedding_summarizer
        → task_multimodal_post_embedding_with_summary   # dict["v3"]
          → task_write_post_embedding_sink_v3
```

**`PlanPostEmbeddingWithSummaryForReply`** (`POST_EMBEDDING_WITH_SUMMARY_FOR_REPLY`) — same spine, reply-only filter and a distinct rate-limit cache.

**`PlanPostEmbeddingV5`** (`MM_EMB_V5`) — no content filter:

```text
task_post_embedding_rate_limit
  → task_media_hydration
    → task_asr_transcription
      → task_multimodal_post_embedding_v5   # dict["v5_1"]
        → task_write_post_embedding_sink_v5   # TaskWriteMMEmbeddingSinkV5SkipKafkaForReplies
```

**`PlanPostEmbeddingV5ForReply`** (`MM_EMB_V5_FOR_REPLY`) inserts `TaskPostEmbeddingWithSummaryForReplyFilter` after the v5-for-reply rate limit, then the same ASR → embed → `TaskWriteMMEmbeddingSinkV5` chain.

</Accordion>
</AccordionGroup>

## Task contract

`Task.exec` is a classmethod with `@retry(stop=stop_after_attempt(2), wait=wait_fixed(1))`. Several classifier/sink tasks override `exec` with `Task.exec.__wrapped__` and skip that retry.

<ParamField body="DISABLE_RULES" type="list[type[DisableTaskRule]]">
Empty by default. First matching `rule.should_disable(ctx)` returns `SKIPPED` before `_exec`.
</ParamField>

| Rule | Skips when |
|---|---|
| `DisableTaskForLocal` | `is_local` |
| `DisableTaskForDev` | `is_dev` |
| `DisableTaskForNonProd` | not `is_prod` |
| `DisableTaskForNonMmEmbProd` | not `is_mm_emb_prod` (embedding sinks / Kafka pub) |
| `DisableTaskForNonPtosProd` | not `is_ptos_prod` (PTOS sink) |

Return / exception map from `Task.exec`:

| Signal | `TaskResultCategory` | Plan DAG |
|---|---|---|
| `should_skip` or `TaskStopExecution` | `SKIPPED` | dependents skip |
| `_exec` completes | `SUCCESS` | dependents run |
| any other exception | re-raised | plan records `ctx.errors` |

Typed wrappers stop with `TaskStopExecution` when the payload is missing the required object:

- `TaskWithPost` — `payload.post`
- `TaskWithUser` — `payload.user`
- `TaskWithUserContext` — `payload.user_context`
- `TaskWithContentAnalysis` — `payload.grox_content_analysis`

`TaskFilter` / `TaskRateLimit` also raise `TaskStopExecution` when `_eligible` is false. Rate-limit caches are process-local `TTLCache(maxsize=10_000, ttl=60)` keyed by post id.

## TaskContext vs TaskResult

`TaskContext` is the per-plan scratch pad. Only a subset is copied onto `TaskResult`.

<ResponseField name="payload" type="TaskPayload">
Inbound payload. Copied onto `TaskResult.task`.
</ResponseField>
<ResponseField name="content_categories" type="list[ContentCategoryResult]">
Copied onto the plan result, then concatenated across plans at merge.
</ResponseField>
<ResponseField name="multimodal_post_embedding" type="list[float] | None">
Copied onto `TaskResult`. Embedding tasks write `multimodal_post_embedding_dict` (`"v3"`, `"v4"`, `"v5_1"`) instead, so this field stays `None` unless something else sets it.
</ResponseField>
<ResponseField name="reason" type="str">
Copied onto `TaskResult.reason`.
</ResponseField>
<ResponseField name="errors" type="list[Exception]">
`success = len(errors) == 0`; `error` is `"\\n".join(str(e) for e in errors)`.
</ResponseField>
<ResponseField name="safety_annotations" type="SafetyPostAnnotations | None">
PTOS only. Consumed by the PTOS sink inside the plan. **Not** a `TaskResult` field.
</ResponseField>
<ResponseField name="reply_ranking_results" type="list[ReplyScoreResult]">
Reply ranking only. Consumed by the Manhattan write task. **Not** a `TaskResult` field.
</ResponseField>
<ResponseField name="summary" type="str">
Used by Kafka publish (`TaskPublishKafka`). **Not** a `TaskResult` field.
</ResponseField>

## ContentClassifier

`ContentClassifier.classify(post)` increments request/intake metrics, calls `_classify`, then records success/error and latency. `_classify` is `_to_convo` → `_sample` → `_parse`.

| Classifier | Categories | Sampler | Parse |
|---|---|---|---|
| `BangerInitialScreenClassifier` | `BANGER_INITIAL_SCREEN`, `GROK_RANKER` | `VisionSampler` / `VLM_PRIMARY`, temperature `0.000001` | `<json>…</json>` → `BangerInitialScreenResult`; `positive = quality_score >= 0.4` |
| `PostSafetyDeluxeClassifier` | `POST_SAFETY_SCREEN` | `VLM_PRIMARY_CRITICAL` | `<json>…</json>` → `tweet_bool_metadata`; `positive=False` |
| `SpamEapiLowFollowerClassifier` | `SPAM_COMMENT` | `VLM_PRIMARY` (overrideable) | JSON `SpamSampleResult`; keeps only `SPAM_COMMENT` and asserts exactly one |
| `SafetyPtosCategoryClassifier` | `SAFETY_PTOS` | `VLM_SAFETY` or deluxe `VLM_PRIMARY_CRITICAL` | Tasks call `classify_post` → `SafetyPostAnnotations`, not `classify` |
| `SafetyPtosPolicyClassifier` | `SAFETY_PTOS` | `VLM_PRIMARY_CRITICAL` (+ EAPI samplers when deluxe) | Per-violation policy prompts: ViolentMedia, AdultContent, Spam, IllegalAndRegulatedBehaviors, HateOrAbuse, ViolentSpeech, SuicideOrSelfHarm |

`ReplyScorer` (`TaskRankReplies`) uses the same VLM stack and `ReplyScoringSystem` prompt but does not subclass `ContentClassifier`.

Banger `classify` optionally takes cached Grok topics from `StratoGrokTopics` (`CACHE_TTL_SECONDS = 3600` on `TaskBangerScreen`).

## TaskResult merge rules

`PlanMaster.merge_results(task, results)` assumes `results` is the non-`None` subset.

| Field | Rule |
|---|---|
| `task` | Original inbound `TaskPayload` |
| `content_categories` | Flatten `r.content_categories` across plans, each `c.model_copy()` |
| `task_started_at` | `min(r.task_started_at)` |
| `task_finished_at` | `max(r.task_finished_at)` |
| `multimodal_post_embedding` | First non-`None` in `ALL_PLANS` order; else `None` |
| `reason` | `"\\n".join` of non-empty `r.reason` |
| `success` | `all(r.success for r in results)` |
| `error` | `"\\n".join` of `r.error or "unknown error"` for every `not r.success` |

<Warning>
If every plan returns `None` (no matching eligibility — including `POST_STREAM_DELAYED`), `merge_results` calls `min`/`max` on an empty sequence and raises. `Engine._run_task` catches that, puts `TaskResult(success=False, error=str(e))`, and increments `engine.task.failed.count`.
</Warning>

Engine failure envelope when `PlanMaster.exec` itself raises:

<ResponseExample>
```python
TaskResult(
    task=task,
    success=False,
    error=str(e),
    task_finished_at=start,          # perf_counter at entry
    task_started_at=time.perf_counter(),  # after the exception
)
```
</ResponseExample>

Dispatcher retry: on `result.success is False` and `task.attempt < dispatcher.max_attempts`, `attempt` is incremented and the same payload is re-queued. Final failure acks the originating generator when `identify_task_origin` resolves.

## Payload fields

<ParamField body="payload_id" type="string" required>
Generator message id (`payload.mid`). Dispatcher inflight set key.
</ParamField>
<ParamField body="eligibilities" type="set[TaskEligibility]">
Default empty. Copied onto `TaskContext.eligibilities`.
</ParamField>
<ParamField body="attempt" type="int">
Default `0`. Incremented by Dispatcher on retry.
</ParamField>
<ParamField body="task_type" type="TaskGeneratorType | None">
Selects deluxe PTOS classifiers/caches when `SAFETY_PTOS_DELUXE`.
</ParamField>
<ParamField body="post / user / user_context / grox_content_analysis" type="optional typed objects">
Defined in unpublished `grox.data_loaders.data_types`. Missing objects skip typed tasks via `TaskStopExecution`.
</ParamField>
<ParamField body="deadline_ts_secs" type="int | None">
Copied from the stream loader; plans do not gate on it.
</ParamField>

## Constraints

- Adding a plan requires an `ALL_PLANS` entry, a unique `REQUIRED_ELIGIBILITY`, and a generator that injects that eligibility.
- `TASK_DEPENDENCIES` must be a closed graph over `TASKS` keys or `Plan.__init__` raises `ValueError`.
- Skip is contagious along outgoing edges; a failed task fails the plan but does not skip siblings already running.
- Intra-plan side effects (Manhattan, Kafka, Strato sinks) read `TaskContext`. Cross-plan merge only carries categories, timestamps, the unused embedding scalar, reason, and success/error.
- `TaskPublishKafka` is disabled in local and dev; UPA / reply-spam / reply-ranking Manhattan writes are disabled outside prod; embedding sinks require `is_mm_emb_prod`; PTOS sink requires `is_ptos_prod`.

## Related pages

<CardGroup>
<Card title="Execute Grox content plans" href="/execute-grox-plans">
Engine, Dispatcher, and GrpcServer startup, fill/result loops, and ack/retry.
</Card>
<Card title="Runtime boundaries" href="/runtime-boundaries">
What this checkout can execute versus unpublished Grox crates and production snapshots.
</Card>
<Card title="Troubleshooting" href="/troubleshooting">
Unpublished Grox modules and other local-run failure modes.
</Card>
</CardGroup>