Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
16 changes: 15 additions & 1 deletion README.md
Original file line number Diff line number Diff line change
Expand Up @@ -11,6 +11,8 @@ checkpointed. Executions can pause, survive process restart, resume, replay, and

The LLM may reason while the graph controls execution.

![StateOps Control Room showing a resolved incident](docs/assets/stateops-control-room.png)

## Why this is a graph rather than a pipeline

The workflow cannot be reduced to a fixed sequence of LLM calls. Investigation expands dynamically
Expand All @@ -34,6 +36,8 @@ application design instead of incidental framework behavior.
- `get_state`, `get_state_history`, replay, and controlled `update_state` forks;
- an atomic Redis `SET NX` boundary for idempotent simulated remediations;
- v3 state streaming, structured logs, and opt-in metadata-only OpenTelemetry;
- provider-portable structured-output schemas with application-enforced collection bounds;
- terminal reasoning failures modeled as serializable `WorkflowError` values and `FAILED` state;
- a production LLM adapter that knows only the provider-neutral Governed LLM Gateway client.

## Engineering evidence map
Expand All @@ -48,6 +52,7 @@ application design instead of incidental framework behavior.
| Typed update-and-route decisions with `Command` | [`VerificationGraph`](src/stateops/graphs/verification/graph.py) |
| History, replay, controlled fork, and nested state reads | [`GraphIncidentRuntime`](src/stateops/graphs/runtime.py) |
| Durable checkpoints and restricted deserialization | [`Redis checkpointer`](src/stateops/adapters/persistence/checkpointer.py) |
| Metadata-safe reasoning failure state | [`graphs/failures.py`](src/stateops/graphs/failures.py) |
| Atomic effect deduplication | [`Redis remediation ledger`](src/stateops/adapters/remediation/redis_executor.py) |
| Real restart/resume proof | [`Redis integration test`](tests/integration/test_redis_runtime.py) |

Expand Down Expand Up @@ -112,6 +117,8 @@ STATEOPS_LLM_WORKLOAD=stateops.incident.reasoning

The external Gateway/Policy Model Router must authorize that dotted workload. StateOps does not
assume that it is authorized and does not fall back to a direct provider when it is denied.
Terminal Gateway failures are reduced to stable error metadata and a `FAILED` phase; raw client or
provider exceptions are never added to checkpoint state.

## Quick start

Expand All @@ -122,7 +129,13 @@ uv sync --frozen --all-groups --extra observability
docker compose up --build
```

Create an incident:
Open [http://127.0.0.1:8000/ui](http://127.0.0.1:8000/ui) to use the **StateOps Control
Room**. The frontend is served by the FastAPI process, uses no CDN or JavaScript framework, and
follows the LangGraph state stream. It can start or load a thread, visualize investigation fan-out,
approve or reject the selected action, and show the verified recovery. **Capture mode** hides the
input controls and expands the dashboard for screenshots.

To drive the same demonstration through the API, create an incident:

```bash
curl --request POST http://127.0.0.1:8000/incidents \
Expand Down Expand Up @@ -187,6 +200,7 @@ dedicated node and the atomic Redis claim makes re-execution safe.
## History, replay, fork, and streaming

```text
GET /ui
GET /incidents/{incident_id}/history
POST /incidents/{incident_id}/replay/{checkpoint_id}
POST /incidents/{incident_id}/fork
Expand Down
18 changes: 17 additions & 1 deletion README.pt-br.md
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,8 @@ processo, ser retomadas, reproduzidas e bifurcadas.

O LLM pode raciocinar enquanto o grafo controla a execução.

![StateOps Control Room mostrando um incidente resolvido](docs/assets/stateops-control-room.png)

## Por que isto é um grafo, e não uma pipeline

O workflow não pode ser reduzido a uma sequência fixa de chamadas a LLMs. A investigação se expande
Expand All @@ -35,6 +37,10 @@ do design da aplicação, em vez de serem comportamentos incidentais do framewor
- `get_state`, `get_state_history`, replay e forks controlados com `update_state`;
- uma barreira Redis atômica com `SET NX` para remediações simuladas idempotentes;
- streaming de estado v3, logs estruturados e OpenTelemetry opcional somente com metadados;
- schemas de saída estruturada portáveis entre providers, com limites de coleção aplicados pela
aplicação;
- falhas terminais de raciocínio modeladas como valores `WorkflowError` serializáveis e estado
`FAILED`;
- um adapter de LLM para produção que conhece apenas o cliente provider-neutral do Governed LLM
Gateway.

Expand All @@ -50,6 +56,7 @@ do design da aplicação, em vez de serem comportamentos incidentais do framewor
| Decisões tipadas de atualização e rota com `Command` | [`VerificationGraph`](src/stateops/graphs/verification/graph.py) |
| Histórico, replay, fork controlado e leitura de estado aninhado | [`GraphIncidentRuntime`](src/stateops/graphs/runtime.py) |
| Checkpoints duráveis e desserialização restrita | [`Checkpointer Redis`](src/stateops/adapters/persistence/checkpointer.py) |
| Estado de falha de raciocínio seguro para metadados | [`graphs/failures.py`](src/stateops/graphs/failures.py) |
| Deduplicação atômica de efeitos | [`Ledger de remediação Redis`](src/stateops/adapters/remediation/redis_executor.py) |
| Prova real de reinicialização e retomada | [`Teste de integração Redis`](tests/integration/test_redis_runtime.py) |

Expand Down Expand Up @@ -115,6 +122,8 @@ STATEOPS_LLM_WORKLOAD=stateops.incident.reasoning
O Gateway/Policy Model Router externo deve autorizar esse workload em formato pontuado. O StateOps não
presume que ele esteja autorizado e não recorre diretamente a um provider quando a solicitação é
negada.
Falhas terminais do Gateway são reduzidas a metadados estáveis de erro e à fase `FAILED`; exceções
brutas do cliente ou do provider nunca são adicionadas ao estado persistido em checkpoint.

## Início rápido

Expand All @@ -125,7 +134,13 @@ uv sync --frozen --all-groups --extra observability
docker compose up --build
```

Crie um incidente:
Abra [http://127.0.0.1:8000/ui](http://127.0.0.1:8000/ui) para usar o **StateOps Control
Room**. O frontend é servido pelo próprio processo FastAPI, não usa CDN nem framework JavaScript e
acompanha o stream de estados do LangGraph. Ele permite iniciar ou carregar uma thread, observar o
fan-out das investigações, aprovar ou rejeitar a ação selecionada e conferir a recuperação. O botão
**Modo captura** remove os controles de entrada e expande o painel para screenshots.

Para operar a mesma demonstração pela API, crie um incidente:

```bash
curl --request POST http://127.0.0.1:8000/incidents \
Expand Down Expand Up @@ -192,6 +207,7 @@ em um nó dedicado, e a reivindicação atômica no Redis torna a reexecução s
## Histórico, replay, fork e streaming

```text
GET /ui
GET /incidents/{incident_id}/history
POST /incidents/{incident_id}/replay/{checkpoint_id}
POST /incidents/{incident_id}/fork
Expand Down
17 changes: 9 additions & 8 deletions docs/DEMO.md
Original file line number Diff line number Diff line change
Expand Up @@ -2,16 +2,17 @@

## Five-minute path

1. Run `docker compose up --build`.
2. Create the checkout incident from the README.
3. Observe three dynamically created investigation branches and `waiting_approval`.
4. Approve `rollback-primary` and observe `resolved`.
5. Read `/incidents/INC-2026-00817/history` and choose the checkpoint whose next node is
1. Run `docker compose up --build` and open `http://127.0.0.1:8000/ui`.
2. Start the prefilled checkout incident in the StateOps Control Room.
3. Observe the dynamically created investigation branches and `waiting_approval`.
4. Capture the interrupted state, approve the selected action, and observe `resolved`.
5. Enable **Capture mode** for a clean screenshot of the resolved graph.
6. Read `/incidents/INC-2026-00817/history` and choose the checkpoint whose next node is
`verification`.
6. Fork it with `selected_action_id=restart-secondary`.
7. Approve the restart. Synthetic verification fails, the graph replans, and a new approval interrupt
7. Fork it with `selected_action_id=restart-secondary`.
8. Approve the restart. Synthetic verification fails, the graph replans, and a new approval interrupt
recommends rollback.
8. Approve rollback. The original idempotency key is detected and the prior simulated effect evidence
9. Approve rollback. The original idempotency key is detected and the prior simulated effect evidence
is reused instead of executing it again.

## Process crash and resume
Expand Down
Binary file added docs/assets/stateops-control-room.png
Loading
Sorry, something went wrong. Reload?
Sorry, we cannot display this file.
Sorry, this file is invalid so it cannot be displayed.
121 changes: 95 additions & 26 deletions src/stateops/adapters/llm/governed_gateway.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,11 @@
from typing import cast

from governed_llm_gateway_client import GatewayClient
from governed_llm_gateway_client.errors import (
GatewayClientError,
GatewayHTTPError,
GatewayTransportError,
)
from governed_llm_gateway_contracts import (
DataClassification,
Message,
Expand All @@ -14,7 +19,9 @@
StructuredOutputSchema,
WorkloadRequirements,
)
from governed_llm_gateway_contracts import ExecutionStatus as GatewayExecutionStatus

from stateops.application.ports.reasoner import IncidentReasoningError
from stateops.domain.enums import ActionKind, Severity
from stateops.domain.models import (
Evidence,
Expand All @@ -26,17 +33,26 @@
)


class GatewayReasoningError(RuntimeError):
class GatewayReasoningError(IncidentReasoningError):
"""Raised when a governed response does not satisfy the StateOps contract."""

def __init__(
self,
message: str,
*,
code: str = "invalid_gateway_response",
retryable: bool = False,
) -> None:
"""Retain stable Gateway metadata without preserving transport objects."""
super().__init__(message, code=code, retryable=retryable)


_HYPOTHESES_SCHEMA: dict[str, object] = {
"type": "object",
"properties": {
"hypotheses": {
"type": "array",
"minItems": 1,
"maxItems": 8,
"items": {
"type": "object",
"properties": {
Expand Down Expand Up @@ -83,13 +99,23 @@ class GatewayReasoningError(RuntimeError):
"actions": {
"type": "array",
"minItems": 1,
"maxItems": 5,
"items": {
"type": "object",
"properties": {
"id": {"type": "string"},
"kind": {"type": "string"},
"parameters": {"type": "object", "additionalProperties": {"type": "string"}},
"parameters": {
"type": "array",
"items": {
"type": "object",
"properties": {
"name": {"type": "string"},
"value": {"type": "string"},
},
"required": ["name", "value"],
"additionalProperties": False,
},
},
"risk": {"type": "string"},
},
"required": ["id", "kind", "parameters", "risk"],
Expand Down Expand Up @@ -131,6 +157,18 @@ def _items(value: object, *, field: str) -> list[Mapping[str, object]]:
return [_mapping(item, field=field) for item in value]


def _bounded_items(
value: object, *, field: str, minimum: int, maximum: int
) -> list[Mapping[str, object]]:
"""Validate collection limits outside provider-specific JSON Schema dialects."""
values = _items(value, field=field)
if not minimum <= len(values) <= maximum:
raise GatewayReasoningError(
f"gateway field {field!r} must contain between {minimum} and {maximum} items"
)
return values


def _text(value: object, *, field: str) -> str:
if not isinstance(value, str) or not value or len(value) > 2000:
raise GatewayReasoningError(f"gateway field {field!r} must be bounded text")
Expand Down Expand Up @@ -179,19 +217,46 @@ async def _generate(
role=MessageRole.USER,
content=f"{instruction}\nINPUT_JSON={json.dumps(payload, separators=(',', ':'))}",
)
response = await self._client.generate(
workload=self._workload,
messages=(message,),
risk_level=RiskLevel.HIGH,
data_classification=DataClassification.INTERNAL,
requirements=WorkloadRequirements(structured_output=True),
structured_output=StructuredOutputSchema(name=schema_name, schema=schema),
context_tokens_estimated=1000,
max_output_tokens=2000,
provider_timeout_seconds=self._timeout_seconds,
)
try:
response = await self._client.generate(
workload=self._workload,
messages=(message,),
risk_level=RiskLevel.HIGH,
data_classification=DataClassification.PUBLIC,
requirements=WorkloadRequirements(structured_output=True),
structured_output=StructuredOutputSchema(name=schema_name, schema=schema),
context_tokens_estimated=1000,
max_output_tokens=2000,
provider_timeout_seconds=self._timeout_seconds,
)
except GatewayHTTPError as exc:
raise GatewayReasoningError(
"gateway rejected the reasoning request",
code=exc.code,
) from exc
except GatewayTransportError as exc:
raise GatewayReasoningError(
"gateway transport failed",
code="gateway_transport_error",
retryable=True,
) from exc
except GatewayClientError as exc:
raise GatewayReasoningError(
"gateway client failed",
code="gateway_client_error",
) from exc
if response.status is not GatewayExecutionStatus.SUCCEEDED:
error = response.error
raise GatewayReasoningError(
"gateway reasoning execution failed",
code=error.code if error is not None else "gateway_execution_failed",
retryable=error.retryable if error is not None else False,
)
if response.content is None:
raise GatewayReasoningError("gateway returned no structured content")
raise GatewayReasoningError(
"gateway returned no structured content",
code="gateway_content_missing",
)
try:
decoded = cast(object, json.loads(response.content))
except json.JSONDecodeError as exc:
Expand All @@ -208,17 +273,18 @@ async def generate_hypotheses(
schema_name="stateops_hypotheses",
schema=_HYPOTHESES_SCHEMA,
)
hypothesis_items = _bounded_items(
payload.get("hypotheses"), field="hypotheses", minimum=1, maximum=8
)
hypotheses = tuple(
Hypothesis(
id=_text(item.get("id"), field="id"),
title=_text(item.get("title"), field="title"),
rationale=_text(item.get("rationale"), field="rationale"),
confidence=_confidence(item.get("confidence")),
)
for item in _items(payload.get("hypotheses"), field="hypotheses")
for item in hypothesis_items
)
if not hypotheses:
raise GatewayReasoningError("gateway returned no hypotheses")
return hypotheses

async def investigate(
Expand Down Expand Up @@ -287,17 +353,22 @@ async def propose_remediations(
schema=_ACTIONS_SCHEMA,
)
actions: list[RemediationAction] = []
for item in _items(payload.get("actions"), field="actions"):
parameters = _mapping(item.get("parameters"), field="parameters")
action_items = _bounded_items(payload.get("actions"), field="actions", minimum=1, maximum=5)
for item in action_items:
parameter_items = _bounded_items(
item.get("parameters"), field="parameters", minimum=0, maximum=16
)
normalized = tuple(
sorted(
(
_text(key, field="parameter_name"),
_text(value, field="parameter_value"),
_text(parameter.get("name"), field="parameter_name"),
_text(parameter.get("value"), field="parameter_value"),
)
for key, value in parameters.items()
for parameter in parameter_items
)
)
if len({name for name, _ in normalized}) != len(normalized):
raise GatewayReasoningError("gateway returned duplicate parameter names")
try:
kind = ActionKind(_text(item.get("kind"), field="kind"))
risk = Severity(_text(item.get("risk"), field="risk"))
Expand All @@ -311,6 +382,4 @@ async def propose_remediations(
risk=risk,
)
)
if not actions:
raise GatewayReasoningError("gateway returned no remediation actions")
return tuple(actions)
10 changes: 10 additions & 0 deletions src/stateops/application/ports/reasoner.py
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,16 @@
)


class IncidentReasoningError(RuntimeError):
"""Metadata-safe failure raised by an incident-reasoning implementation."""

def __init__(self, message: str, *, code: str, retryable: bool = False) -> None:
"""Retain only stable failure metadata that is safe to checkpoint."""
super().__init__(message)
self.code = code
self.retryable = retryable


class IncidentReasoner(Protocol):
"""Reason about incidents without exposing model or provider selection."""

Expand Down
Loading