# Skill: Add Custom State Fields

Use when the built-in `PipelineState` fields are not enough for your pipeline's data needs.
Read `arccrew/orchestrator/state.py` fully before extending.

## When to Extend vs Use Existing Fields

`PipelineState` already provides these fields for free:

| Field | Type | Reducer | Use for |
|-------|------|---------|---------|
| `context` | `dict` | `_take_last_dict` | Intermediate data passed between agents |
| `result` | `dict` | `_take_last_dict` | Final output of the current task |
| `results` | `list[dict]` | `operator.add` | Accumulated results across tasks |
| `verification_result` | `dict` | `_take_last_dict` | Verifier agent output |
| `review_result` | `dict` | `_take_last_dict` | Reviewer agent output |
| `retry_count` | `int` | `_take_last` | Current retry attempt number |
| `retry_history` | `list[dict]` | (none) | History of retry attempts |
| `messages` | `list` | `add_messages` | LangChain message history |
| `tasks` | `list[dict]` | (none) | Input tasks list |
| `current_task_index` | `int` | `_take_last` | Which task is being processed |

**Use `context` first.** Most pipelines can fit all inter-agent data in `context` as a dict.
If your agents pass structured data between them and it fits in a `dict`, use `context`.

**Extend PipelineState when:**
- You need a list that accumulates across multiple agents (reducer: `operator.add`)
- You need a field that is conceptually separate from `context` for code clarity
- You need type safety — TypedDict fields are validated
- You have domain objects (campaigns, products, leads) that are first-class concepts

**Do NOT extend for:**
- Simple string or number values that fit in `context` (just use `context["my_key"]`)
- Temporary scratch space within a single agent (use local variables)
- Configuration that belongs in `.env` / `get_settings()`

## The 4 Reducer Choices

LangGraph reducers resolve conflicts when multiple nodes write to the same field.
The reducer runs every time a node returns an update for that field.

### `_take_last` — Keep most recent value

```python
current_task_index: Annotated[int, _take_last]
```

Use for: integers, booleans, simple scalars. The most recent write wins.
If a node does not update this field, the previous value is kept.

### `_take_last_dict` — Keep most recent dict

```python
context: Annotated[dict, _take_last_dict]
strategy: Annotated[dict, _take_last_dict]
```

Use for: dicts representing current state, not history. Each agent replaces the previous value.
Fallback to `{}` if never set.

### `_take_last_str` — Keep most recent string

```python
current_agent: Annotated[str, _take_last_str]
campaign_id: Annotated[str, _take_last_str]
```

Use for: string identifiers, labels, status strings. Fallback to `""` if never set.

### `operator.add` — Accumulate (append list items)

```python
results: Annotated[list[dict], operator.add]
campaigns_processed: Annotated[list[str], operator.add]
```

Use for: accumulating items across tasks or agents. Each update appends to the list.
Every write must be a list (even single items: `[item]`).

### No Reducer (bare field)

```python
tasks: list[dict]
batch_id: str
```

Use for: fields set once at pipeline start and never updated by agents.
If two nodes write to this field, the last one wins (no merging).

## Creating `my_project/state.py`

```python
"""Domain-specific state for the my_project pipeline."""

import operator
from typing import Annotated

from arccrew.orchestrator.state import PipelineState, _take_last_dict, _take_last_str


class MyState(PipelineState):
    """Extended state for my_project pipeline.

    Adds domain-specific fields on top of PipelineState.
    Inherits: messages, tasks, current_task_index, context, result, results,
              verification_result, review_result, retry_count, retry_history,
              batch_id, start_time, dry_run.
    """

    # Example: a dict field that gets replaced each time (strategy, config, etc.)
    strategy: Annotated[dict, _take_last_dict]

    # Example: an accumulator — each agent appends to this
    processed_items: Annotated[list[dict], operator.add]

    # Example: a string identifier that changes per-task
    current_campaign_id: Annotated[str, _take_last_str]

    # Example: a plain field set once at startup
    project_name: str
```

Import from `arccrew.orchestrator.state` — do not copy the reducer functions.

## Using MyState in `pipeline.py`

Replace `PipelineState` with your custom class wherever the graph is created:

```python
from langgraph.graph import END, START, StateGraph
from state import MyState  # instead of PipelineState

def create_pipeline():
    ...
    graph = StateGraph(MyState)   # <-- use MyState here
    ...
    return graph.compile()
```

Also use `MyState` in `create_pipeline()` factory from `arccrew/orchestrator/graph.py`:

```python
from arccrew.orchestrator.graph import create_pipeline as make_graph
from state import MyState

graph = make_graph(
    state_class=MyState,   # <-- pass MyState
    nodes={...},
    flow=[...],
)
```

## Reading State Fields in `execute()`

Access any field with `.get()` and a sensible default:

```python
@track_timing
async def execute(self, state: dict) -> Command:
    # PipelineState fields
    tasks = state.get("tasks", [])
    idx = state.get("current_task_index", 0)
    context = state.get("context", {})

    # Custom MyState fields
    strategy = state.get("strategy", {})
    campaign_id = state.get("current_campaign_id", "")
    project_name = state.get("project_name", "unknown")

    ...
```

Note: even though `MyState` is a TypedDict, `execute()` receives a plain `dict` at runtime.
Use `.get()` with defaults to be safe.

## Returning State Updates via `Command`

Return only the fields you are updating — LangGraph merges them:

```python
return Command(
    goto="next_node",
    update={
        "context": {"processed": True, "data": result_data},  # replaces context
        "processed_items": [{"id": campaign_id, "status": "done"}],  # appended via operator.add
        "current_campaign_id": next_campaign_id,  # replaces
    }
)
```

For `operator.add` fields, always wrap the item(s) in a list:
```python
update={"processed_items": [single_item]}   # correct
update={"processed_items": single_item}      # WRONG — not a list, will error
```

## Building the Initial State

When invoking the pipeline (in tests or from `pipeline.py`), include your custom fields
in the initial state dict. Fields not included start as `None` (or the TypedDict default).

```python
initial_state = {
    # PipelineState fields
    "messages": [],
    "tasks": [{"description": "..."}],
    "current_task_index": 0,
    "current_agent": "init",
    "context": {},
    "result": {},
    "verification_result": {},
    "review_result": {},
    "results": [],
    "retry_count": 0,
    "retry_history": [],
    "batch_id": "run-001",
    "start_time": datetime.now(),
    "dry_run": False,

    # MyState custom fields
    "strategy": {"mode": "aggressive"},
    "processed_items": [],
    "current_campaign_id": "",
    "project_name": "my_project",
}
```

## Full Example: Adding a `campaigns` Accumulator

Problem: a pipeline processes multiple campaigns and needs each agent to append its output
to a running list, not overwrite it.

```python
# my_project/state.py
import operator
from typing import Annotated
from arccrew.orchestrator.state import PipelineState

class CampaignState(PipelineState):
    campaigns_output: Annotated[list[dict], operator.add]
```

```python
# In the agent's execute()
return Command(
    goto="next_agent",
    update={
        "campaigns_output": [{"campaign_id": "c123", "score": 0.87, "recommendation": "..."}],
    }
)
```

After all tasks complete, `state["campaigns_output"]` contains every agent's output in order.
Access it in a final summarizer agent:
```python
all_outputs = state.get("campaigns_output", [])
```
