Add durable mid-run steering, safe-point input injection, and pending-rerun semantics

CHANGELOG.md

@@ -1,5 +1,70 @@
 # Changelog
 
+## Unreleased
+
+### Durable mid-run steering (issue #16)
+
+- 新增 `POST /v1/runs/{run_id}/steer`：在 Run 执行期间（含 `paused`）durably 注入新的
+  结构化输入，不等待完成、不强制取消、也不需要先暂停。支持 `Idempotency-Key`
+  header/body 去重、`kind`/`payload`/`metadata` 结构化 payload、大小上限校验。
+- Python SDK（同步/异步）新增 `client.runs.steer(run_id, kind=..., payload=..., metadata=...,
+  idempotency_key=...)`。
+- Runtime 新增 `runtime.has_steering`、`runtime.peek_steering()`、
+  `runtime.drain_steering()`；`compiled_graph.steer(run_id, ...)` 在嵌入式/无 Server
+  场景下提供同一套语义，未使用时零开销、不泄漏 channel 状态。
+- Run 生命周期新增 `run.steer.accepted`、`run.steer.consumed`、`run.steer.superseded`
+  SSE 事件；`consumed` 事件携带 `steering_event_id`、`source_event_id`、`sequence`、
+  `kind`、`queue_latency_seconds`、`node`、`namespace`、`task_id`；`superseded` 事件
+  在 resume-transfer 与 final-boundary disposition 两种场景下幂等写入，同样携带
+  `sequence`/`kind`/`source_event_id`。
+- Resume 语义：`POST /v1/runs/{run_id}/resume` 在**同一原子事务**中创建新 Run 并把旧
+  Run 上仍 `pending`/`delivered` 的 steering 事件迁移到新 Run（保序、保留身份）；旧
+  Run 之后再收到 `/steer` 或 `/cancel` 返回 `409 run_superseded`。`source_event_id`
+  在跨多次 pause/resume 时始终指向客户端最初 `/steer` 收到的根 id，`created_at`/
+  `queue_latency_seconds` 从最初 durable accepted 时刻起算，不因中间的 resume 跳数
+  被重置。旧 Run 是否已被 resume 取代由新增的 typed `runs.superseded_by_run_id` 列
+  记录（与该事务同时原子写入），而不是用户可写的 `metadata`，避免调用方在创建 Run
+  时伪造该字段来假冒 lineage 状态。
+- Cancel 语义：`request_cancel()` 现在对每个非终态状态都有明确策略——`pending`/
+  `paused`（没有存活 worker 可协作）立即变为 `cancelled`；`running` 变为
+  `cancelling`，由持有该 Run 的 worker 协作完成终结；已被 resume 取代
+  （`superseded_by_run_id` 已设置）的历史 Run 拒绝为 `409 run_superseded`，不会
+  假成功。`claim_run()` 的租约过期回收同样区分：过期的 `running` 恢复为 `pending`；
+  过期的 `cancelling`（worker 在收到 cancel 后、终结提交前崩溃）恢复为 `cancelled`，
+  避免 Run 永久卡死在 `cancelling` 并阻塞同 thread 后续任务。
+- `commit_steering_consumptions_if_owned()`（状态更新 + `run.steer.consumed` 写入）
+  与 `finalize_run_with_steering_disposition_if_owned()`（终态提交 + 未消费 steering
+  的 `superseded` disposition + `run.steer.superseded` 写入）均在同一事务/临界区内
+  完成，并按 `(lease_owner, attempt)` fenced：同一批 consumption 的重复提交（例如
+  worker 在收到 ack 前连接断开后重试）只会产生一次 `run.steer.consumed`；旧 attempt
+  在 lease 被新 worker 接管后无法再修改该 Run 的 durable 状态。`retry_run_with_event_
+  if_owned()` 同样把状态转换与 `worker_retrying` 事件合并为一次原子写入，cancel 优先
+  于 stale 的 retry 结果。
+- PostgreSQL schema：新增 `run_steering_events` 表及 `runs` 表的多个新增列，共 4 个
+  纯新增、forward-only 的 Alembic revision：
+  - `0002_steering` — `run_steering_events` 表本身；
+  - `0003_steering_source_event` — `source_event_id` 列/局部索引；
+  - `0004_steering_closed` — `runs.steering_closed`（finalization 阶段的 steering
+    admission 门控）；
+  - `0005_superseded_by_run_id` — `runs.superseded_by_run_id`（typed resume-lineage
+    marker，见上）。
+  `alembic upgrade head` 是生产环境唯一支持的迁移路径，**必须到达
+  `0005_superseded_by_run_id`**——`/steer`、`/cancel`、finalization、resume 的生产
+  SQL 直接依赖 `0004`/`0005` 新增的列，只迁移到 `0003` 会导致这些代码路径在缺列的
+  schema 上报 `UndefinedColumn` 错误。四个 revision 均可在部署新二进制前对存活数据库
+  提前执行（schema 层面向后兼容，旧二进制会忽略新增的 nullable 列）；但 schema
+  兼容不等于语义兼容——只有在控制面实例全部升级到本版本后，`steering_closed`/
+  `superseded_by_run_id` 才会被读取并生效，回滚二进制会失去这些新语义（不是完整的
+  语义回滚保证），必要时应在回滚前 quiesce 或等待 in-flight Run 结束。Redis 仅用于
+  可选的低延迟通知，从不作为唯一数据源；Redis 丢失不影响最终投递。
+- 兼容性：不影响不使用 steering 的现有图与调用方；上述 schema 变更均为纯新增列/表，
+  无 breaking API 变更。
+- 新增中英文文档：`Wiki/content/docs/{zh,en}/api/threads-runs.mdx`（steer API 全量
+  语义）、`python-sdk.mdx`（SDK 示例）、`errors-events.mdx`（错误码与生命周期事件）、
+  新增概念页 `Wiki/content/docs/{zh,en}/concepts/steering.mdx`、新增运维页
+  `Wiki/content/docs/{zh,en}/operations/steering-operations.mdx`；`docs/api.md`
+  同步更新。
+
 ## 2.2.0
 
 ### Cache-first prompt optimization

README.en.md

@@ -55,6 +55,7 @@ flowchart TB
 LingxiGraph is for product and platform teams that need to deliver Agent experiences reliably to end users. It focuses on the engineering problems that appear after a model call works in a prototype:
 
 - **Continuous user experiences** — threads, checkpoints, history, forks, resumes, and resumable SSE support multi-turn sessions, long-running work, and reconnects.
+- **Durable mid-run steering** — inject new structured input into a running (or paused) run without cancelling it; delivery is durable, ordered, deduplicated, and survives worker crashes and paused-run resumes.
 - **Composable intelligent workflows** — state graphs, subgraphs, and supervisor, handoff, swarm, group chat, plan-and-execute, parallel-review, and map-reduce patterns organize complex requests.
 - **Governed production execution** — pinned graph versions, lease-based queues, idempotency keys, timeouts, budgets, quotas, dead-letter/redrive, and cooperative cancellation make retries and scaling manageable.
 - **Tenant-aware security boundaries** — OIDC/JWT, RBAC, tenant isolation, PostgreSQL RLS, audit records, and safe JSON state serialization provide runtime foundations for data isolation in online products.

README.md

@@ -32,6 +32,8 @@ StateGraph · Runtime · Checkpoint · Events
 
 - **StateGraph**：用节点和边描述 Agent 与业务执行逻辑。
 - **Durable Execution**：支持 checkpoint、暂停恢复、重试与长任务执行。
+- **Durable Steering**：运行中（甚至暂停中）durably 注入新的结构化输入，无需取消 run；投递
+  durable、有序、去重，并能在 worker 崩溃和 paused run resume 后依然可靠送达。
 - **Streaming Events**：通过 Python Runtime、REST 与可续传 SSE 暴露运行状态。
 - **Production Runtime**：提供 Agent Server、Worker、持久化、权限、审计与观测能力。
 - **Open Integrations**：支持 OpenAI-compatible、Coze、A2A、MCP 与开放 Agent Skills。

Wiki/content/docs/en/api/errors-events.mdx

@@ -29,6 +29,16 @@ Non-2xx responses use `application/problem+json`:
 
 Common stable codes include `idempotency_conflict`, `quota_exceeded`, and `join_timeout`, plus problem codes mapped from schema, authentication, concurrency, and resource-state failures. Treat new codes as an unknown error rather than failing to parse them.
 
+### Steering-specific error codes
+
+| `code` | HTTP | Triggered when |
+| --- | --- | --- |
+| `run_terminal` | 409 | `/steer` is called against a run in a terminal status (`succeeded`/`failed`/`cancelled`/`timed_out`/`dead_letter`); no durable write is created |
+| `run_superseded` | 409 | `/steer` or `/cancel` is called against an old `run_id` already superseded by a `resume`; call it against the new run's `id` instead (see `metadata.resumed_from_run_id`). For `/cancel`, this fires even though the old run's `status` is still `paused` -- the check reads the typed `superseded_by_run_id` column, not `status` and not user-writable `metadata`, so the historical run is never mutated to `cancelled` while its resumed descendant keeps running, and the check cannot be forged via a client-submitted `metadata` key (see the operations doc's "The `superseded_by_run_id` lineage marker") |
+| `payload_too_large` | 413/422 | Serialized `payload` + `metadata` exceed the size cap (32KB, or the server's event size cap, whichever is smaller); no durable write is created |
+| `run_resume_conflict` | 409 | Concurrent resume calls raced against the same paused run; exactly one wins and creates the new run, every other caller gets this instead |
+| `run_finalizing` | 409 | `/steer` is called after the run's graph has already finished executing but before its final status is committed (a short window while the worker durably flushes and finalizes); there is no safe point left to consume the event, so it is rejected rather than durably accepted with nobody to ever read it. A `paused` outcome is not affected -- steering a paused run remains normal |
+
 ## Run business errors
 
 Successful creation returns `202 pending`. A later node/schema/provider error moves the resource to `failed`, `timed_out`, or `dead_letter` and populates:
@@ -67,6 +77,73 @@ data: {"id":"...","run_id":"...","sequence":18,"kind":"node_completed","data":{}
 - Lines beginning `: heartbeat` are keep-alive comments and should be ignored.
 - The server closes the stream when the run becomes terminal or `paused`.
 
+### Steering lifecycle events
+
+Besides ordinary execution events (`node_completed`, etc.), mid-run input (steering)
+produces three dedicated SSE event kinds across its lifecycle:
+
+| `kind` | Emitted when | Key fields |
+| --- | --- | --- |
+| `run.steer.accepted` | A `/steer` call is durably accepted (202) | `steering_event_id`, `source_event_id` (if from a resume migration), `sequence`, `kind` |
+| `run.steer.consumed` | Graph code calls `drain_steering()` and that consumption is durably committed by `commit_steering_consumptions_if_owned()` | `steering_event_id`, `source_event_id`, `sequence`, `kind`, `queue_latency_seconds`, `node`, `namespace`, `task_id` |
+| `run.steer.superseded` | A steering event's row durably transitions to the `superseded` disposition without ever being consumed — either because a paused run it belonged to was resumed (its still-pending steering is migrated onto the new run) or because it was still `pending`/`delivered` when the run reached its terminal-finalization boundary (see below) | `steering_event_id`, `source_event_id`, `sequence`, `kind`, `reason` (`resume_transfer` or `unconsumed_at_final_boundary`), `superseded_by_run_id` (set for `resume_transfer`, `null` for `unconsumed_at_final_boundary`), `replacement_steering_event_id` (set for `resume_transfer`, `null` for `unconsumed_at_final_boundary`) |
+
+`run.steer.superseded`'s `unconsumed_at_final_boundary` reason is a deliberately honest,
+generic label — it does **not** claim the event was racing anything. It covers every steering
+row still `pending`/`delivered` once the owning worker reaches a genuinely terminal
+finalization boundary, and that has more than one possible cause: a `/steer` call can durably
+commit its row a moment before the worker closes new admission (a real race), but just as
+commonly the graph simply never called `drain_steering()` at all for a row that had been
+sitting there, fully visible, for the run's entire lifetime — ordinary application-level
+non-consumption, not a race. The Runtime cannot currently distinguish these two cases from the
+data it has, so it does not pretend to; both get the same `unconsumed_at_final_boundary`
+reason. Rather than forcing the run through an indefinite retry loop just because a steering
+event happened to exist — which would penalize a graph that legitimately never calls
+`drain_steering()` at all — the Runtime durably transitions the row to `superseded` and
+finalizes the run with its already-computed outcome. A `run.steer.superseded` event is never
+silently dropped in this case; it is always paired with the disposition change, committed
+together with the run's terminal status in the same durable transaction, and is idempotent
+under retry (a repeated attempt to supersede an already-superseded row emits no duplicate
+event).
+
+Field meanings:
+
+- `steering_event_id`: the id of the steering event under the **current** run (if the event
+  was migrated from an older run, this is the fresh id it got after migration, not the id
+  the client originally received);
+- `source_event_id`: the **root** id the client originally received from its `/steer` call.
+  Even across repeated pause/resume migrations of the same underlying steering, this field
+  stays consistent at every hop, letting a client correlate the eventual
+  `run.steer.consumed`/`run.steer.superseded` back to its original request. This is
+  per-lifecycle-event, not a blanket rule: on `run.steer.accepted` and `run.steer.consumed` it
+  is present only for events that arrived via a migration — a directly-submitted event (never
+  transferred) has no `source_event_id` on those two. `run.steer.superseded` is different: it
+  always sets `source_event_id` to `event.source_event_id or event.id`, so a directly-submitted
+  event's superseded record still carries a root id — its own — even though it was never
+  migrated;
+- `sequence`: this event's position in its run's steering sequence (a separate counter from
+  the run's overall event-stream `sequence`);
+- `kind`: passed through from the `/steer` request body's `kind`;
+- `queue_latency_seconds`: seconds from the steering event's **original** durably-accepted
+  time (including time accumulated across every resume) to when it was consumed;
+- `node`: the name of the node that consumed the event;
+- `namespace`: the subgraph namespace path (array) the consuming node belongs to;
+- `task_id`: distinguishes the specific task among parallel/`Send` tasks in the same
+  superstep.
+
+**PostgreSQL is the durable source of truth; Redis is only an optional accelerator.** All three
+of `run.steer.accepted`, `run.steer.consumed`, and `run.steer.superseded` are durably written
+to PostgreSQL before becoming visible anywhere else. Redis pub/sub is used only to hint
+workers that "steering may be waiting," reducing polling latency. Even if that Redis
+notification is lost (a dropped connection, a Redis restart), a worker still checks/polls
+PostgreSQL at its next safe point and eventually consumes the event — a lost Redis
+notification never permanently drops steering. Similarly, the production Worker path,
+`commit_steering_consumptions_if_owned()`, is idempotent: when a worker retries the same
+consumption batch after a transient failure, only events that actually transition status
+produce a new `run.steer.consumed` — already-consumed events are never re-emitted. The same
+holds for superseding a stale row: retrying an already-superseded transition emits no
+duplicate `run.steer.superseded`.
+
 ## Retry policy
 
 1. Use bounded exponential backoff only for `retryable=true`, network failure, 429, and transient 5xx.

Wiki/content/docs/en/api/overview.mdx

@@ -59,6 +59,8 @@ stateDiagram-v2
   running --> timed_out
   running --> cancelling
   cancelling --> cancelled
+  pending --> cancelled: cancel (no live worker)
+  paused --> cancelled: cancel (no live worker)
   failed --> pending: redrive
   dead_letter --> pending: redrive
   paused --> pending: resume creates new run

Wiki/content/docs/en/api/python-sdk.mdx

@@ -36,6 +36,16 @@ with LingxiGraphClient(
 
     completed = client.runs.join(run["id"], timeout=120)
     print(completed["output"])
+
+    # Durably inject mid-run input; success only means it was durably
+    # written -- not that the graph has processed it yet.
+    accepted = client.runs.steer(
+        run["id"],
+        kind="user_input",
+        payload={"message": "switch to Spanish"},
+        idempotency_key="msg-123",
+    )
+    print(accepted["id"], accepted["status"])  # "pending"
 ```
 
 Available resource methods:
@@ -45,12 +55,35 @@ Available resource methods:
 | `graphs` | `list()`, `get(graph_id)` |
 | `assistants` | `create(**values)`, `list()`, `get()`, `update()`, `delete()` |
 | `threads` | `create()`, `list()`, `get()`, `update()`, `state()`, `history()`, `fork()`, `delete()` |
-| `runs` | `create()`, `get()`, `list(thread_id)`, `join()`, `stream()`, `cancel()`, `resume()` |
+| `runs` | `create()`, `get()`, `list(thread_id)`, `join()`, `stream()`, `cancel()`, `steer()`, `resume()` |
 | `store` | `batch(operations)`, `search(namespace, ...)` |
 | `schedules` | `create()`, `list()`, `update()`, `delete()` |
 
 Use `client.request(method, path, **kwargs)` for redrive or an endpoint not yet wrapped by a resource.
 
+### `runs.steer()`
+
+```python
+client.runs.steer(
+    run_id,
+    kind="user_input",
+    payload={"message": "..."},
+    metadata=None,
+    idempotency_key="msg-123",
+)
+```
+
+- `kind` defaults to `"user_input"` but may be any application-defined string;
+- `payload` is the raw JSON the graph reads back via `runtime.drain_steering()`;
+- `idempotency_key`, when set, is also sent as the `Idempotency-Key` header; a repeat call
+  with the same key returns the original event instead of enqueueing a second one;
+- the return value is the 202 response body (`id`, `sequence`, `status`) — it does **not**
+  mean the graph has processed the input; only a subsequent `run.steer.consumed` event
+  means it was actually consumed;
+- calling it against a terminal run raises `LingxiGraphAPIError` with
+  `code == "run_terminal"`; calling it against an old `run_id` already superseded by a
+  `resume` raises `code == "run_superseded"`.
+
 ## Asynchronous client
 
 ```python
@@ -65,6 +98,14 @@ async with AsyncLingxiGraphClient(base_url, token=token) as client:
     )
     async for event in client.runs.stream(run["id"]):
         print(event)
+
+    accepted = await client.runs.steer(
+        run["id"],
+        kind="user_input",
+        payload={"message": "switch to Spanish"},
+        idempotency_key="msg-123",
+    )
+    print(accepted["id"], accepted["status"])
 ```
 
 Async resources mirror sync resources. Do not use the synchronous client inside an async event loop.