mirror of
https://github.com/instructkr/claw-code.git
synced 2026-10-14 00:00:19 +08:00
Compare commits
9
Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
6d7412d958 | ||
|
|
479e598cd3 | ||
|
|
89e717ec09 | ||
|
|
01e9fc3229 | ||
|
|
02c22b4211 | ||
|
|
0e54dec1c9 | ||
|
|
c4ce758d5e | ||
|
|
4a920fb281 | ||
|
|
192d40a860 |
@@ -5,6 +5,56 @@ All notable changes to this project will be documented in this file.
|
||||
The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
|
||||
and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
|
||||
|
||||
## [0.1.6] - 2026-10-13
|
||||
|
||||
### Added
|
||||
|
||||
- SDK session lifecycle management with session reconciliation and recovery
|
||||
|
||||
### Fixed
|
||||
|
||||
- Session lifecycle outcome serialization and preservation in recovery workflows
|
||||
|
||||
## [0.1.5] - 2026-10-13
|
||||
|
||||
### Added
|
||||
|
||||
- CLI turn interruption: press Esc to interrupt running turns and cancel in-flight model requests
|
||||
- Custom OpenAI provider namespace with separate environment variables for multi-tenant scenarios
|
||||
- `/tools` and `/info` slash commands for command discovery and runtime information
|
||||
- Team enhancements: subagentModel accessor chain and parallel tool execution support
|
||||
- Configuration options for concurrent tool calls and parallel subagent execution (`maxParallelToolCalls`, `maxConcurrentAgents`)
|
||||
- Per-team subagent concurrency control via AgentGate
|
||||
- SDK operation registry with schema validation for extensible operation handling
|
||||
- Bundled skills registry and skill chaining support for skill composition
|
||||
- Per-session settings override with schema validation
|
||||
- WebSocket client transport completion for SDK wire protocol support
|
||||
|
||||
### Fixed
|
||||
|
||||
- Turn interruption signal handling for graceful cancellation of in-flight requests
|
||||
- Markdown streaming block spacing in CLI output
|
||||
- Spinner restoration for proper is_terminal() check in run_turn
|
||||
- Model token limit catch-all to return None for unknown models instead of panicking
|
||||
- Parallel tool result ordering by call index with fan-out capping
|
||||
- Workspace unsafe_code=forbid restoration across all crates
|
||||
- TaskClaim TOCTOU (time-of-check-time-of-use) vulnerability
|
||||
- TUI dependency removal and claw-rag-service compilation on non-Linux platforms
|
||||
- Clippy lint violations in trident, compact, and claw-rag-service crates
|
||||
- PermissionMode comparison to use explicit .covers() method instead of direct operators
|
||||
- OPENAI_BASE_URL environment isolation in provider hint tests
|
||||
|
||||
### Changed
|
||||
|
||||
- Enhanced permission mode checking for consistency and clarity
|
||||
- Improved team and concurrent agent configuration schema
|
||||
- Refactored slash command processing for /tools and /info implementations
|
||||
|
||||
### Known Limitations
|
||||
|
||||
- ACP / Zed integration is not yet implemented (status stub available via `claw acp`)
|
||||
- Python porting workspace is for reference only, not used in production
|
||||
|
||||
## [0.1.4] - 2026-10-13
|
||||
|
||||
### Added
|
||||
|
||||
+4
-4
@@ -87,11 +87,11 @@
|
||||
</p>
|
||||
|
||||
<p align="center">
|
||||
<a href="https://star-history.com/#ultraworkers/claw-code&Date">
|
||||
<a href="https://star-history.dera.page/#ultraworkers/claw-code&Date">
|
||||
<picture>
|
||||
<source media="(prefers-color-scheme: dark)" srcset="https://api.star-history.com/svg?repos=ultraworkers/claw-code&type=Date&theme=dark" />
|
||||
<source media="(prefers-color-scheme: light)" srcset="https://api.star-history.com/svg?repos=ultraworkers/claw-code&type=Date" />
|
||||
<img alt="Star history for ultraworkers/claw-code" src="https://api.star-history.com/svg?repos=ultraworkers/claw-code&type=Date" width="600" />
|
||||
<source media="(prefers-color-scheme: dark)" srcset="https://star-history.dera.page/svg?repos=ultraworkers/claw-code&type=Date&theme=dark" />
|
||||
<source media="(prefers-color-scheme: light)" srcset="https://star-history.dera.page/svg?repos=ultraworkers/claw-code&type=Date" />
|
||||
<img alt="Star history for ultraworkers/claw-code" src="https://star-history.dera.page/svg?repos=ultraworkers/claw-code&type=Date" width="600" />
|
||||
</picture>
|
||||
</a>
|
||||
</p>
|
||||
|
||||
@@ -161,52 +161,52 @@ GJC sessions have **three entry points**:
|
||||
|
||||
### claw-code Status
|
||||
|
||||
**Current** (as of PR #3330): Partial. SDK transport merged (see SDK Wire Protocol). Local session lifecycle exists in `runtime::Session`. Remaining: operation handlers, session router with durable lookup, invocation reconciliation store, terminal scope tracking.
|
||||
**Current**: The SDK now exposes `sdk::session::SessionRouter` with durable `turn.create`, `turn.prompt`, `turn.steer`, `turn.abort_and_prompt`, and paginated `turn.list` handling. It bridges `runtime::Session`/`ConversationRuntime`, streams ordered lifecycle events, persists invocation outcomes for reconciliation, and stores terminal session scope by correlation key. The WebSocket transport remains a client transport; a remotely hosted broker and automatic recovery of an unknown post-crash outcome remain outside this slice.
|
||||
|
||||
**Files**:
|
||||
- `rust/crates/runtime/src/session.rs` (1,961 LOC)
|
||||
- `rust/crates/runtime/src/conversation.rs` (1,878 LOC)
|
||||
- `rust/crates/runtime/src/conversation.rs` (conversation loop and lifecycle event bridge)
|
||||
- `rust/crates/runtime/src/session_control.rs` (file ops, migration)
|
||||
- `rust/crates/runtime/src/lib.rs` (re-exports `Session`, `SessionStore`, `ConversationRuntime`)
|
||||
- `rust/crates/sdk/src/session/` (router, event stream, reconciliation, terminal scope)
|
||||
|
||||
**Lines**: ~3,900 LOC (session + conversation)
|
||||
|
||||
**Reality**:
|
||||
- `Session` struct holds in-memory state (messages, usage, context).
|
||||
- `SessionStore` reads/writes `.claw/sessions/{id}/` directory.
|
||||
- `ConversationRuntime` drives the local agent loop (sync).
|
||||
- No WebSocket or remote access.
|
||||
- No invocation reconciliation or uncertain outcome recovery.
|
||||
- No terminal scope tracking (per-session-per-correlation).
|
||||
- `Session` holds persisted conversation state and `SessionStore` resolves session IDs in the workspace namespace.
|
||||
- `ConversationRuntime` drives the local agent loop and emits lifecycle events to an optional callback.
|
||||
- `sdk::session::SessionRouter` dispatches the five turn operations and exposes a per-invocation event stream.
|
||||
- Invocation terminal outcomes and correlation-key terminal scopes are durable; startup recovery marks unfinished invocations uncertain for later reconciliation.
|
||||
- A remote WebSocket server/broker and inference of an outcome that was never durably recorded are still absent.
|
||||
|
||||
### Gap Assessment
|
||||
|
||||
| Component | GJC | claw-code | Gap | Complexity |
|
||||
|-----------|-----|-----------|-----|------------|
|
||||
| Remote Session Router | 2,500 LOC | — | 100% | High |
|
||||
| Invocation Reconciliation | 1,200 LOC | — | 100% | High |
|
||||
| Terminal Scope Tracking | 600 LOC | — | 100% | Medium |
|
||||
| Async Event Streaming | 800 LOC | Partial | 60% | Medium |
|
||||
| Uncertain Outcome Recovery | 1,000 LOC | — | 100% | High |
|
||||
| **Total** | **6,100 LOC** | **3,900** | **2,200 LOC** | — |
|
||||
| Remote Session Router | 2,500 LOC | Local router/dispatch | Partial | High |
|
||||
| Invocation Reconciliation | 1,200 LOC | Durable invocation records | Partial | High |
|
||||
| Terminal Scope Tracking | 600 LOC | Implemented | — | Medium |
|
||||
| Async Event Streaming | 800 LOC | Local per-invocation stream | Partial | Medium |
|
||||
| Uncertain Outcome Recovery | 1,000 LOC | Unfinished records recovered as uncertain | Partial | High |
|
||||
| **Total** | **6,100 LOC** | **Session lifecycle slice implemented** | **Remote broker remains** | — |
|
||||
|
||||
### Proposed Implementation
|
||||
|
||||
**Owner Crate**: `sdk` (new, or extend `runtime`)
|
||||
|
||||
**Module Structure** (add to `sdk/session/`):
|
||||
**Module Structure** (`sdk/session/`):
|
||||
```
|
||||
sdk/session/
|
||||
├── router.rs (session routing, durable lookup)
|
||||
├── mod.rs (session routing and operation dispatch)
|
||||
├── reconciliation.rs (uncertain → terminal recovery)
|
||||
├── terminal_scope.rs (per-correlation state retention)
|
||||
└── events.rs (event publishing, reconciliation)
|
||||
└── events.rs (ordered lifecycle event types)
|
||||
```
|
||||
|
||||
**Integration Points**:
|
||||
- Bridge `runtime::Session` into `sdk::Session` wrapper
|
||||
- Add `runtime::SessionStore` extensions for reconciliation data
|
||||
- Extend `ConversationRuntime` with event emission
|
||||
- Bridge `runtime::Session` and `ConversationRuntime` into SDK lifecycle operations
|
||||
- Persist reconciliation and terminal-scope data alongside the session namespace
|
||||
- Emit lifecycle callbacks from `ConversationRuntime`
|
||||
|
||||
**Size Estimate**: 2,500 LOC (Rust, new + modifications)
|
||||
|
||||
|
||||
Generated
+13
-11
@@ -93,7 +93,7 @@ checksum = "7f202df86484c868dbad7eaa557ef785d5c66295e41b460ef922eca0723b842c"
|
||||
|
||||
[[package]]
|
||||
name = "api"
|
||||
version = "0.1.4"
|
||||
version = "0.1.6"
|
||||
dependencies = [
|
||||
"criterion",
|
||||
"reqwest",
|
||||
@@ -478,7 +478,7 @@ checksum = "c8d4a3bb8b1e0c1050499d1815f5ab16d04f0959b233085fb31653fbfc9d98f9"
|
||||
|
||||
[[package]]
|
||||
name = "claw-analog"
|
||||
version = "0.1.4"
|
||||
version = "0.1.6"
|
||||
dependencies = [
|
||||
"api",
|
||||
"clap",
|
||||
@@ -498,7 +498,7 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "claw-rag-service"
|
||||
version = "0.1.4"
|
||||
version = "0.1.6"
|
||||
dependencies = [
|
||||
"axum 0.8.9",
|
||||
"blake3",
|
||||
@@ -531,7 +531,7 @@ checksum = "1d07550c9036bf2ae0c684c4297d503f838287c83c53686d05370d0e139ae570"
|
||||
|
||||
[[package]]
|
||||
name = "commands"
|
||||
version = "0.1.4"
|
||||
version = "0.1.6"
|
||||
dependencies = [
|
||||
"plugins",
|
||||
"runtime",
|
||||
@@ -540,7 +540,7 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "compat-harness"
|
||||
version = "0.1.4"
|
||||
version = "0.1.6"
|
||||
dependencies = [
|
||||
"commands",
|
||||
"runtime",
|
||||
@@ -1568,7 +1568,7 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "mock-anthropic-service"
|
||||
version = "0.1.4"
|
||||
version = "0.1.6"
|
||||
dependencies = [
|
||||
"api",
|
||||
"serde_json",
|
||||
@@ -1761,7 +1761,7 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "plugins"
|
||||
version = "0.1.4"
|
||||
version = "0.1.6"
|
||||
dependencies = [
|
||||
"serde",
|
||||
"serde_json",
|
||||
@@ -2136,7 +2136,7 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "runtime"
|
||||
version = "0.1.4"
|
||||
version = "0.1.6"
|
||||
dependencies = [
|
||||
"glob",
|
||||
"plugins",
|
||||
@@ -2261,7 +2261,7 @@ checksum = "b39cdef0fa800fc44525c84ccb54a029961a8215f9619753635a9c0d2538d46d"
|
||||
|
||||
[[package]]
|
||||
name = "rusty-claude-cli"
|
||||
version = "0.1.4"
|
||||
version = "0.1.6"
|
||||
dependencies = [
|
||||
"api",
|
||||
"commands",
|
||||
@@ -2337,8 +2337,10 @@ name = "sdk"
|
||||
version = "0.1.4"
|
||||
dependencies = [
|
||||
"futures-util",
|
||||
"runtime",
|
||||
"serde",
|
||||
"serde_json",
|
||||
"tempfile",
|
||||
"tokio",
|
||||
"tokio-tungstenite",
|
||||
"tracing",
|
||||
@@ -2619,7 +2621,7 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "telemetry"
|
||||
version = "0.1.4"
|
||||
version = "0.1.6"
|
||||
dependencies = [
|
||||
"serde",
|
||||
"serde_json",
|
||||
@@ -2899,7 +2901,7 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "tools"
|
||||
version = "0.1.4"
|
||||
version = "0.1.6"
|
||||
dependencies = [
|
||||
"api",
|
||||
"aspect-core",
|
||||
|
||||
+1
-1
@@ -3,7 +3,7 @@ members = ["crates/*"]
|
||||
resolver = "2"
|
||||
|
||||
[workspace.package]
|
||||
version = "0.1.4"
|
||||
version = "0.1.6"
|
||||
edition = "2021"
|
||||
license = "MIT"
|
||||
publish = false
|
||||
|
||||
@@ -305,6 +305,11 @@ mod tests {
|
||||
|
||||
#[test]
|
||||
fn provider_detection_prefers_model_family() {
|
||||
let _lock = env_lock();
|
||||
let _anthropic = EnvVarGuard::set("ANTHROPIC_API_KEY", None);
|
||||
let _openai = EnvVarGuard::set("OPENAI_API_KEY", None);
|
||||
let _xai = EnvVarGuard::set("XAI_API_KEY", None);
|
||||
|
||||
assert_eq!(detect_provider_kind("grok-3"), ProviderKind::Xai);
|
||||
assert_eq!(
|
||||
detect_provider_kind("claude-sonnet-4-6"),
|
||||
|
||||
@@ -82,7 +82,7 @@ async fn send_message_posts_json_and_parses_response() {
|
||||
);
|
||||
assert_eq!(
|
||||
request.headers.get("user-agent").map(String::as_str),
|
||||
Some("claude-code/0.1.4")
|
||||
Some("claude-code/0.1.6")
|
||||
);
|
||||
assert_eq!(
|
||||
request.headers.get("anthropic-beta").map(String::as_str),
|
||||
|
||||
File diff suppressed because it is too large
Load Diff
@@ -84,8 +84,8 @@ pub use config_validate::{
|
||||
};
|
||||
pub use conversation::{
|
||||
auto_compaction_threshold_from_env, ApiClient, ApiRequest, AssistantEvent, AutoCompactionEvent,
|
||||
ConversationRuntime, PromptCacheEvent, RuntimeError, StaticToolExecutor, ToolError,
|
||||
ToolExecutor, TurnInterruptSignal, TurnSummary,
|
||||
ConversationEvent, ConversationRuntime, PromptCacheEvent, RuntimeError, StaticToolExecutor,
|
||||
ToolError, ToolExecutor, TurnControlSignal, TurnInterruptSignal, TurnSummary,
|
||||
};
|
||||
pub use file_ops::{
|
||||
edit_file, edit_file_in_workspace, glob_search, glob_search_in_workspace, grep_search,
|
||||
|
||||
@@ -453,7 +453,7 @@ fn direct_resume_safe_slash_commands_route_to_local_json_actions_831() {
|
||||
|
||||
for (command, expected_kind, expected_status) in [
|
||||
("/version", "version", "ok"),
|
||||
("/sandbox", "sandbox", "warn"),
|
||||
("/sandbox", "sandbox", "ok"),
|
||||
("/diff", "diff", "ok"),
|
||||
("/status", "status", "ok"),
|
||||
] {
|
||||
@@ -470,10 +470,19 @@ fn direct_resume_safe_slash_commands_route_to_local_json_actions_831() {
|
||||
.unwrap_or_else(|_| panic!("{command} must emit JSON (#831), got: {stdout:?}"));
|
||||
|
||||
assert_eq!(parsed["kind"], expected_kind, "{command} kind: {parsed}");
|
||||
assert_eq!(
|
||||
parsed["status"], expected_status,
|
||||
"{command} status: {parsed}"
|
||||
);
|
||||
// /sandbox status varies by environment: "ok" when namespace isolation is available,
|
||||
// "warn" when running in restricted environments (e.g., CI without unshare support)
|
||||
if command == "/sandbox" {
|
||||
assert!(
|
||||
parsed["status"] == "ok" || parsed["status"] == "warn",
|
||||
"{command} status should be 'ok' or 'warn' but got: {parsed}"
|
||||
);
|
||||
} else {
|
||||
assert_eq!(
|
||||
parsed["status"], expected_status,
|
||||
"{command} status: {parsed}"
|
||||
);
|
||||
}
|
||||
assert_ne!(
|
||||
parsed["error_kind"], "interactive_only",
|
||||
"{command} must not emit interactive_only (#831): {parsed}"
|
||||
|
||||
@@ -13,3 +13,10 @@ serde = { version = "1", features = ["derive"] }
|
||||
uuid = { version = "1", features = ["v4", "serde"] }
|
||||
tracing = "0.1"
|
||||
url = "2"
|
||||
runtime = { path = "../runtime" }
|
||||
|
||||
[dev-dependencies]
|
||||
tempfile = "3"
|
||||
|
||||
[lints]
|
||||
workspace = true
|
||||
|
||||
@@ -15,6 +15,8 @@ pub enum SdkErrorCode {
|
||||
Unauthorized,
|
||||
EndpointCredentialForbidden,
|
||||
Unavailable,
|
||||
/// Requested session state conflicts with an active invocation
|
||||
Conflict,
|
||||
ConnectionError,
|
||||
ConnectionClosed,
|
||||
Timeout,
|
||||
@@ -37,6 +39,7 @@ impl SdkErrorCode {
|
||||
Self::Unauthorized => "unauthorized",
|
||||
Self::EndpointCredentialForbidden => "endpoint_credential_forbidden",
|
||||
Self::Unavailable => "unavailable",
|
||||
Self::Conflict => "conflict",
|
||||
Self::ConnectionError => "connection_error",
|
||||
Self::ConnectionClosed => "connection_closed",
|
||||
Self::Timeout => "timeout",
|
||||
@@ -59,6 +62,7 @@ impl SdkErrorCode {
|
||||
"unauthorized" => Self::Unauthorized,
|
||||
"endpoint_credential_forbidden" => Self::EndpointCredentialForbidden,
|
||||
"unavailable" => Self::Unavailable,
|
||||
"conflict" => Self::Conflict,
|
||||
"connection_error" => Self::ConnectionError,
|
||||
"connection_closed" => Self::ConnectionClosed,
|
||||
"timeout" => Self::Timeout,
|
||||
@@ -132,6 +136,14 @@ impl SdkError {
|
||||
Self::new(SdkErrorCode::Unavailable, message)
|
||||
}
|
||||
|
||||
pub fn not_found(message: impl Into<String>) -> Self {
|
||||
Self::new(SdkErrorCode::NotFound, message)
|
||||
}
|
||||
|
||||
pub fn conflict(message: impl Into<String>) -> Self {
|
||||
Self::new(SdkErrorCode::Conflict, message)
|
||||
}
|
||||
|
||||
pub fn connection_error(message: impl Into<String>) -> Self {
|
||||
Self::new(SdkErrorCode::ConnectionError, message)
|
||||
}
|
||||
@@ -209,6 +221,13 @@ mod tests {
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn conflict_error_code_round_trips_on_the_wire() {
|
||||
let error = SdkError::conflict("session is active");
|
||||
assert_eq!(error.code.to_string(), "conflict");
|
||||
assert_eq!(SdkErrorCode::from_wire("conflict"), SdkErrorCode::Conflict);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn unknown_remote_code_is_preserved_with_typed_internal_code() {
|
||||
let error = SdkError::from_remote("future_error", "failed");
|
||||
|
||||
@@ -14,9 +14,16 @@
|
||||
pub mod error;
|
||||
pub mod frame;
|
||||
pub mod registry;
|
||||
pub mod session;
|
||||
pub mod transport;
|
||||
|
||||
pub use error::{SdkError, SdkErrorCode, SdkResult, SdkSentRecord};
|
||||
pub use frame::{FrameType, SdkFrame};
|
||||
pub use registry::{OperationMetadata, OperationRegistry, Schema};
|
||||
pub use session::{
|
||||
InvocationOutcome, InvocationReconciliationStore, InvocationRecord, InvocationStatus,
|
||||
RuntimeSessionRunner, SessionDispatch, SessionEvent, SessionEventStream, SessionEventType,
|
||||
SessionInvocation, SessionPage, SessionRouter, SessionRunResult, SessionSummary,
|
||||
SessionTurnRunner, TerminalScopeRecord, TerminalScopeStore,
|
||||
};
|
||||
pub use transport::{SdkClient, SdkClientConfig};
|
||||
|
||||
@@ -263,6 +263,89 @@ impl OperationRegistry {
|
||||
)),
|
||||
);
|
||||
|
||||
// SDK session lifecycle operations.
|
||||
for (name, description, required) in [
|
||||
(
|
||||
"turn.create",
|
||||
"Create a durable SDK session",
|
||||
Vec::<&str>::new(),
|
||||
),
|
||||
(
|
||||
"turn.prompt",
|
||||
"Prompt a session and stream lifecycle events",
|
||||
Vec::<&str>::new(),
|
||||
),
|
||||
(
|
||||
"turn.steer",
|
||||
"Steer an active session invocation",
|
||||
vec!["invocationId"],
|
||||
),
|
||||
(
|
||||
"turn.abort_and_prompt",
|
||||
"Abort the active invocation and start a replacement prompt",
|
||||
Vec::<&str>::new(),
|
||||
),
|
||||
(
|
||||
"turn.list",
|
||||
"List persisted SDK sessions",
|
||||
Vec::<&str>::new(),
|
||||
),
|
||||
] {
|
||||
let mut properties = serde_json::Map::new();
|
||||
properties.insert("sessionId".to_string(), json!({ "type": "string" }));
|
||||
properties.insert("correlationKey".to_string(), json!({ "type": "string" }));
|
||||
properties.insert("invocationId".to_string(), json!({ "type": "string" }));
|
||||
properties.insert("prompt".to_string(), json!({ "type": "string" }));
|
||||
properties.insert(
|
||||
"override".to_string(),
|
||||
json!({ "type": ["string", "object", "array"] }),
|
||||
);
|
||||
properties.insert(
|
||||
"request".to_string(),
|
||||
json!({ "type": ["string", "object", "array"] }),
|
||||
);
|
||||
properties.insert(
|
||||
"input".to_string(),
|
||||
json!({ "type": ["string", "object", "array"] }),
|
||||
);
|
||||
properties.insert("tools".to_string(), json!({ "type": "array" }));
|
||||
properties.insert("settings".to_string(), json!({ "type": "object" }));
|
||||
properties.insert("model".to_string(), json!({ "type": "string" }));
|
||||
properties.insert("cursor".to_string(), json!({ "type": "string" }));
|
||||
properties.insert(
|
||||
"limit".to_string(),
|
||||
json!({ "type": "integer", "minimum": 1 }),
|
||||
);
|
||||
let mut input_schema = json!({
|
||||
"type": "object",
|
||||
"properties": properties,
|
||||
"required": required,
|
||||
});
|
||||
let prompt_alternatives = json!([
|
||||
{ "required": ["prompt"] },
|
||||
{ "required": ["request"] },
|
||||
{ "required": ["input"] },
|
||||
]);
|
||||
match name {
|
||||
"turn.prompt" | "turn.abort_and_prompt" => {
|
||||
input_schema["anyOf"] = prompt_alternatives;
|
||||
}
|
||||
"turn.steer" => {
|
||||
input_schema["anyOf"] = json!([
|
||||
{ "required": ["override"] },
|
||||
{ "required": ["prompt"] },
|
||||
{ "required": ["request"] },
|
||||
{ "required": ["input"] },
|
||||
]);
|
||||
}
|
||||
_ => {}
|
||||
}
|
||||
registry.register(
|
||||
OperationMetadata::new(name, description)
|
||||
.with_input_schema(Schema::new(format!("Input for {name}"), input_schema)),
|
||||
);
|
||||
}
|
||||
|
||||
registry
|
||||
}
|
||||
|
||||
@@ -397,6 +480,20 @@ mod tests {
|
||||
assert!(registry.has_operation("events.replay"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn standard_registry_has_session_lifecycle_operations() {
|
||||
let registry = OperationRegistry::with_standard_operations();
|
||||
for operation in [
|
||||
"turn.create",
|
||||
"turn.prompt",
|
||||
"turn.steer",
|
||||
"turn.abort_and_prompt",
|
||||
"turn.list",
|
||||
] {
|
||||
assert!(registry.has_operation(operation), "missing {operation}");
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn registry_validation_catches_missing_required_fields() {
|
||||
let registry = OperationRegistry::with_standard_operations();
|
||||
|
||||
@@ -0,0 +1,31 @@
|
||||
//! Event stream types for session lifecycle operations.
|
||||
|
||||
use serde::{Deserialize, Serialize};
|
||||
use serde_json::Value;
|
||||
|
||||
/// Lifecycle events published for one SDK turn invocation.
|
||||
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
|
||||
#[serde(rename_all = "snake_case")]
|
||||
pub enum SessionEventType {
|
||||
TurnAccepted,
|
||||
AgentBegin,
|
||||
ToolUse,
|
||||
ToolResult,
|
||||
AgentEnd,
|
||||
TurnSteered,
|
||||
TurnEnded,
|
||||
}
|
||||
|
||||
/// One ordered session lifecycle event.
|
||||
#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
|
||||
#[serde(rename_all = "camelCase")]
|
||||
pub struct SessionEvent {
|
||||
pub event_id: String,
|
||||
pub event_type: SessionEventType,
|
||||
pub session_id: String,
|
||||
pub invocation_id: String,
|
||||
pub sequence: u64,
|
||||
pub timestamp_ms: u64,
|
||||
pub durable: bool,
|
||||
pub payload: Value,
|
||||
}
|
||||
File diff suppressed because it is too large
Load Diff
@@ -0,0 +1,673 @@
|
||||
//! Durable invocation state for recovering SDK sessions across process restarts.
|
||||
|
||||
use std::{
|
||||
fs::{self, OpenOptions},
|
||||
io::Write,
|
||||
path::{Path, PathBuf},
|
||||
sync::Mutex,
|
||||
time::{SystemTime, UNIX_EPOCH},
|
||||
};
|
||||
|
||||
#[cfg(unix)]
|
||||
use std::fs::File;
|
||||
|
||||
use serde::{Deserialize, Serialize};
|
||||
use uuid::Uuid;
|
||||
|
||||
use crate::error::{SdkError, SdkResult};
|
||||
|
||||
static RECONCILIATION_WRITE_LOCK: Mutex<()> = Mutex::new(());
|
||||
|
||||
/// Lifecycle state persisted for an invocation.
|
||||
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
|
||||
#[serde(rename_all = "snake_case")]
|
||||
pub enum InvocationStatus {
|
||||
Accepted,
|
||||
Running,
|
||||
Uncertain,
|
||||
Terminal,
|
||||
}
|
||||
|
||||
/// Result captured when an invocation reaches a terminal state.
|
||||
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
|
||||
#[serde(rename_all = "snake_case")]
|
||||
pub enum InvocationOutcome {
|
||||
Completed {
|
||||
message_count: usize,
|
||||
},
|
||||
Aborted {
|
||||
message_count: usize,
|
||||
},
|
||||
Failed {
|
||||
message: String,
|
||||
message_count: usize,
|
||||
},
|
||||
}
|
||||
|
||||
/// Persisted lifecycle and reconciliation data for one invocation.
|
||||
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
|
||||
pub struct InvocationRecord {
|
||||
pub invocation_id: String,
|
||||
pub session_id: String,
|
||||
pub correlation_key: Option<String>,
|
||||
pub status: InvocationStatus,
|
||||
pub accepted_at_ms: u64,
|
||||
pub updated_at_ms: u64,
|
||||
pub outcome: Option<InvocationOutcome>,
|
||||
pub uncertainty_reason: Option<String>,
|
||||
pub terminal_event: Option<serde_json::Value>,
|
||||
pub terminal_scope_revision: Option<u64>,
|
||||
}
|
||||
|
||||
/// File-backed store for invocation lifecycle reconciliation.
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct InvocationReconciliationStore {
|
||||
root: PathBuf,
|
||||
}
|
||||
|
||||
impl InvocationReconciliationStore {
|
||||
/// Creates a store handle without creating its root directory.
|
||||
#[must_use]
|
||||
pub fn new(root: impl Into<PathBuf>) -> Self {
|
||||
Self { root: root.into() }
|
||||
}
|
||||
|
||||
/// Inserts a new invocation record. Existing IDs are not overwritten.
|
||||
pub fn insert(&self, record: InvocationRecord) -> SdkResult<()> {
|
||||
let _guard = RECONCILIATION_WRITE_LOCK
|
||||
.lock()
|
||||
.unwrap_or_else(std::sync::PoisonError::into_inner);
|
||||
let path = self.record_path(&record.invocation_id)?;
|
||||
if self.get(&record.invocation_id)?.is_some() {
|
||||
return Err(SdkError::invalid_input(format!(
|
||||
"invocation '{}' already exists",
|
||||
record.invocation_id
|
||||
)));
|
||||
}
|
||||
atomic_write_json(&path, &record)
|
||||
}
|
||||
|
||||
/// Marks a non-terminal invocation as running.
|
||||
pub fn mark_running(&self, invocation_id: &str) -> SdkResult<()> {
|
||||
let _guard = RECONCILIATION_WRITE_LOCK
|
||||
.lock()
|
||||
.unwrap_or_else(std::sync::PoisonError::into_inner);
|
||||
let mut record = self.required_record(invocation_id)?;
|
||||
if record.status != InvocationStatus::Accepted {
|
||||
return Err(SdkError::conflict(format!(
|
||||
"invocation '{}' cannot start from state {:?}",
|
||||
invocation_id, record.status
|
||||
)));
|
||||
}
|
||||
record.status = InvocationStatus::Running;
|
||||
record.uncertainty_reason = None;
|
||||
record.updated_at_ms = updated_at_ms(record.updated_at_ms);
|
||||
self.write_record(&record)
|
||||
}
|
||||
|
||||
/// Marks a non-terminal invocation as uncertain with a recovery reason.
|
||||
pub fn mark_uncertain(&self, invocation_id: &str, reason: impl Into<String>) -> SdkResult<()> {
|
||||
let reason = reason.into();
|
||||
self.update_nonterminal(invocation_id, move |record| {
|
||||
record.status = InvocationStatus::Uncertain;
|
||||
record.uncertainty_reason = Some(reason);
|
||||
})
|
||||
}
|
||||
|
||||
/// Records a terminal result. Once terminal, a record cannot be changed again.
|
||||
pub fn record_terminal(
|
||||
&self,
|
||||
invocation_id: &str,
|
||||
outcome: InvocationOutcome,
|
||||
terminal_event: Option<serde_json::Value>,
|
||||
) -> SdkResult<()> {
|
||||
self.record_terminal_with_scope_revision(invocation_id, outcome, terminal_event, None)
|
||||
}
|
||||
|
||||
/// Records a terminal result together with its durable correlation-scope revision.
|
||||
pub fn record_terminal_with_scope_revision(
|
||||
&self,
|
||||
invocation_id: &str,
|
||||
outcome: InvocationOutcome,
|
||||
terminal_event: Option<serde_json::Value>,
|
||||
terminal_scope_revision: Option<u64>,
|
||||
) -> SdkResult<()> {
|
||||
let _guard = RECONCILIATION_WRITE_LOCK
|
||||
.lock()
|
||||
.unwrap_or_else(std::sync::PoisonError::into_inner);
|
||||
let mut record = self.required_record(invocation_id)?;
|
||||
if record.status == InvocationStatus::Terminal {
|
||||
return Err(SdkError::invalid_input(format!(
|
||||
"invocation '{}' is already terminal",
|
||||
invocation_id
|
||||
)));
|
||||
}
|
||||
if record
|
||||
.outcome
|
||||
.as_ref()
|
||||
.is_some_and(|pending| pending != &outcome)
|
||||
|| record
|
||||
.terminal_event
|
||||
.as_ref()
|
||||
.is_some_and(|pending| Some(pending) != terminal_event.as_ref())
|
||||
{
|
||||
return Err(SdkError::conflict(format!(
|
||||
"invocation '{}' has a different pending terminal result",
|
||||
invocation_id
|
||||
)));
|
||||
}
|
||||
record.status = InvocationStatus::Terminal;
|
||||
record.outcome = Some(outcome);
|
||||
record.uncertainty_reason = None;
|
||||
record.terminal_event = terminal_event;
|
||||
record.terminal_scope_revision = terminal_scope_revision;
|
||||
record.updated_at_ms = updated_at_ms(record.updated_at_ms);
|
||||
self.write_record(&record)
|
||||
}
|
||||
|
||||
/// Persists a known turn result before its correlation scope is updated.
|
||||
///
|
||||
/// The record remains non-terminal until the scope update succeeds, allowing
|
||||
/// startup or a later reconciliation read to safely finish the transition.
|
||||
pub fn record_pending_terminal(
|
||||
&self,
|
||||
invocation_id: &str,
|
||||
outcome: InvocationOutcome,
|
||||
terminal_event: serde_json::Value,
|
||||
) -> SdkResult<()> {
|
||||
let _guard = RECONCILIATION_WRITE_LOCK
|
||||
.lock()
|
||||
.unwrap_or_else(std::sync::PoisonError::into_inner);
|
||||
let mut record = self.required_record(invocation_id)?;
|
||||
if record.status == InvocationStatus::Terminal {
|
||||
return Err(SdkError::invalid_input(format!(
|
||||
"invocation '{}' is already terminal",
|
||||
invocation_id
|
||||
)));
|
||||
}
|
||||
if record.outcome.is_some() || record.terminal_event.is_some() {
|
||||
return Err(SdkError::conflict(format!(
|
||||
"invocation '{}' already has a pending terminal result",
|
||||
invocation_id
|
||||
)));
|
||||
}
|
||||
record.outcome = Some(outcome);
|
||||
record.terminal_event = Some(terminal_event);
|
||||
record.terminal_scope_revision = None;
|
||||
record.updated_at_ms = updated_at_ms(record.updated_at_ms);
|
||||
self.write_record(&record)
|
||||
}
|
||||
|
||||
/// Loads one invocation record, if present.
|
||||
pub fn get(&self, invocation_id: &str) -> SdkResult<Option<InvocationRecord>> {
|
||||
let path = self.record_path(invocation_id)?;
|
||||
read_record(&path, invocation_id)
|
||||
}
|
||||
|
||||
/// Loads all invocation records in stable invocation-ID order.
|
||||
pub fn list(&self) -> SdkResult<Vec<InvocationRecord>> {
|
||||
let entries = match fs::read_dir(&self.root) {
|
||||
Ok(entries) => entries,
|
||||
Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(Vec::new()),
|
||||
Err(error) => {
|
||||
return Err(io_error(
|
||||
"read invocation store directory",
|
||||
&self.root,
|
||||
error,
|
||||
))
|
||||
}
|
||||
};
|
||||
|
||||
let mut records = Vec::new();
|
||||
for entry in entries {
|
||||
let entry = entry
|
||||
.map_err(|error| io_error("read invocation store entry", &self.root, error))?;
|
||||
if !entry
|
||||
.file_type()
|
||||
.map_err(|error| io_error("inspect invocation store entry", &entry.path(), error))?
|
||||
.is_file()
|
||||
{
|
||||
continue;
|
||||
}
|
||||
|
||||
let path = entry.path();
|
||||
if path.extension() != Some(std::ffi::OsStr::new("json")) {
|
||||
continue;
|
||||
}
|
||||
let invocation_id = path
|
||||
.file_stem()
|
||||
.and_then(|stem| stem.to_str())
|
||||
.ok_or_else(|| SdkError::invalid_input("invocation filename is not valid UTF-8"))?;
|
||||
validate_invocation_id(invocation_id)?;
|
||||
if let Some(record) = read_record(&path, invocation_id)? {
|
||||
records.push(record);
|
||||
}
|
||||
}
|
||||
records.sort_by(|left, right| left.invocation_id.cmp(&right.invocation_id));
|
||||
Ok(records)
|
||||
}
|
||||
|
||||
/// Loads records associated with the exact correlation key.
|
||||
pub fn list_by_correlation_key(&self, key: &str) -> SdkResult<Vec<InvocationRecord>> {
|
||||
Ok(self
|
||||
.list()?
|
||||
.into_iter()
|
||||
.filter(|record| record.correlation_key.as_deref() == Some(key))
|
||||
.collect())
|
||||
}
|
||||
|
||||
/// Durably marks accepted and running records uncertain after a restart.
|
||||
///
|
||||
/// Returns the records changed by this recovery pass; already uncertain and
|
||||
/// terminal records are left untouched and are not included in the result.
|
||||
pub fn reconcile_unfinished(&self) -> SdkResult<Vec<InvocationRecord>> {
|
||||
let _guard = RECONCILIATION_WRITE_LOCK
|
||||
.lock()
|
||||
.unwrap_or_else(std::sync::PoisonError::into_inner);
|
||||
let mut recovered = Vec::new();
|
||||
for mut record in self.list()? {
|
||||
if matches!(
|
||||
record.status,
|
||||
InvocationStatus::Accepted | InvocationStatus::Running
|
||||
) {
|
||||
record.status = InvocationStatus::Uncertain;
|
||||
record.uncertainty_reason =
|
||||
Some("recovered unfinished invocation after restart".into());
|
||||
record.updated_at_ms = updated_at_ms(record.updated_at_ms);
|
||||
self.write_record(&record)?;
|
||||
recovered.push(record);
|
||||
}
|
||||
}
|
||||
Ok(recovered)
|
||||
}
|
||||
|
||||
fn record_path(&self, invocation_id: &str) -> SdkResult<PathBuf> {
|
||||
validate_invocation_id(invocation_id)?;
|
||||
Ok(self.root.join(format!("{invocation_id}.json")))
|
||||
}
|
||||
|
||||
fn required_record(&self, invocation_id: &str) -> SdkResult<InvocationRecord> {
|
||||
self.get(invocation_id)?.ok_or_else(|| {
|
||||
SdkError::invalid_input(format!("invocation '{invocation_id}' does not exist"))
|
||||
})
|
||||
}
|
||||
|
||||
fn update_nonterminal(
|
||||
&self,
|
||||
invocation_id: &str,
|
||||
update: impl FnOnce(&mut InvocationRecord),
|
||||
) -> SdkResult<()> {
|
||||
let _guard = RECONCILIATION_WRITE_LOCK
|
||||
.lock()
|
||||
.unwrap_or_else(std::sync::PoisonError::into_inner);
|
||||
let mut record = self.required_record(invocation_id)?;
|
||||
if record.status == InvocationStatus::Terminal {
|
||||
return Err(SdkError::invalid_input(format!(
|
||||
"invocation '{}' is already terminal",
|
||||
invocation_id
|
||||
)));
|
||||
}
|
||||
update(&mut record);
|
||||
record.updated_at_ms = updated_at_ms(record.updated_at_ms);
|
||||
self.write_record(&record)
|
||||
}
|
||||
|
||||
fn write_record(&self, record: &InvocationRecord) -> SdkResult<()> {
|
||||
let path = self.record_path(&record.invocation_id)?;
|
||||
atomic_write_json(&path, record)
|
||||
}
|
||||
}
|
||||
|
||||
/// Atomically serializes a value beside its target and renames it into place.
|
||||
pub(super) fn atomic_write_json<T: Serialize + ?Sized>(path: &Path, value: &T) -> SdkResult<()> {
|
||||
let bytes = serde_json::to_vec(value)
|
||||
.map_err(|error| SdkError::internal(format!("serialize {}: {error}", path.display())))?;
|
||||
let parent = path
|
||||
.parent()
|
||||
.filter(|parent| !parent.as_os_str().is_empty())
|
||||
.unwrap_or_else(|| Path::new("."));
|
||||
fs::create_dir_all(parent)
|
||||
.map_err(|error| io_error("create store directory", parent, error))?;
|
||||
let filename = path
|
||||
.file_name()
|
||||
.ok_or_else(|| SdkError::invalid_input("store path has no filename"))?;
|
||||
let temporary_path = parent.join(format!(
|
||||
".{}.{}.tmp",
|
||||
filename.to_string_lossy(),
|
||||
Uuid::new_v4()
|
||||
));
|
||||
|
||||
let write_result = (|| -> std::io::Result<()> {
|
||||
let mut temporary = OpenOptions::new()
|
||||
.write(true)
|
||||
.create_new(true)
|
||||
.open(&temporary_path)?;
|
||||
temporary.write_all(&bytes)?;
|
||||
temporary.sync_all()?;
|
||||
drop(temporary);
|
||||
fs::rename(&temporary_path, path)?;
|
||||
#[cfg(unix)]
|
||||
File::open(parent)?.sync_all()?;
|
||||
Ok(())
|
||||
})();
|
||||
|
||||
match write_result {
|
||||
Ok(()) => Ok(()),
|
||||
Err(error) => {
|
||||
let _ = fs::remove_file(&temporary_path);
|
||||
Err(io_error("atomically write JSON", path, error))
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn read_record(path: &Path, expected_id: &str) -> SdkResult<Option<InvocationRecord>> {
|
||||
let bytes = match fs::read(path) {
|
||||
Ok(bytes) => bytes,
|
||||
Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(None),
|
||||
Err(error) => return Err(io_error("read invocation record", path, error)),
|
||||
};
|
||||
let record: InvocationRecord = serde_json::from_slice(&bytes).map_err(|error| {
|
||||
SdkError::internal(format!(
|
||||
"parse invocation record {}: {error}",
|
||||
path.display()
|
||||
))
|
||||
})?;
|
||||
validate_invocation_id(&record.invocation_id)?;
|
||||
if record.invocation_id != expected_id {
|
||||
return Err(SdkError::internal(format!(
|
||||
"invocation record ID '{}' does not match filename ID '{expected_id}'",
|
||||
record.invocation_id
|
||||
)));
|
||||
}
|
||||
Ok(Some(record))
|
||||
}
|
||||
|
||||
fn validate_invocation_id(invocation_id: &str) -> SdkResult<()> {
|
||||
if invocation_id.is_empty()
|
||||
|| invocation_id.len() > 200
|
||||
|| !invocation_id
|
||||
.bytes()
|
||||
.all(|byte| byte.is_ascii_alphanumeric() || byte == b'-' || byte == b'_')
|
||||
{
|
||||
return Err(SdkError::invalid_input(format!(
|
||||
"unsafe invocation ID '{invocation_id}'"
|
||||
)));
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
fn updated_at_ms(previous: u64) -> u64 {
|
||||
let now = SystemTime::now()
|
||||
.duration_since(UNIX_EPOCH)
|
||||
.unwrap_or_default()
|
||||
.as_millis()
|
||||
.min(u128::from(u64::MAX)) as u64;
|
||||
now.max(previous.saturating_add(1))
|
||||
}
|
||||
|
||||
fn io_error(action: &str, path: &Path, error: std::io::Error) -> SdkError {
|
||||
SdkError::internal(format!("{action} '{}': {error}", path.display()))
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
struct TestDirectory(PathBuf);
|
||||
|
||||
impl TestDirectory {
|
||||
fn new() -> Self {
|
||||
let path = std::env::temp_dir().join(format!("sdk-session-{}", Uuid::new_v4()));
|
||||
fs::create_dir(&path).expect("create test directory");
|
||||
Self(path)
|
||||
}
|
||||
}
|
||||
|
||||
impl Drop for TestDirectory {
|
||||
fn drop(&mut self) {
|
||||
let _ = fs::remove_dir_all(&self.0);
|
||||
}
|
||||
}
|
||||
|
||||
fn accepted_record(invocation_id: &str) -> InvocationRecord {
|
||||
InvocationRecord {
|
||||
invocation_id: invocation_id.into(),
|
||||
session_id: "session-a".into(),
|
||||
correlation_key: Some("request-a".into()),
|
||||
status: InvocationStatus::Accepted,
|
||||
accepted_at_ms: 10,
|
||||
updated_at_ms: 10,
|
||||
outcome: None,
|
||||
uncertainty_reason: None,
|
||||
terminal_event: None,
|
||||
terminal_scope_revision: None,
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn constructor_has_no_filesystem_side_effects() {
|
||||
let root = std::env::temp_dir().join(format!("sdk-session-{}", Uuid::new_v4()));
|
||||
let _store = InvocationReconciliationStore::new(&root);
|
||||
assert!(!root.exists());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn invocation_records_survive_store_restart() {
|
||||
let root = TestDirectory::new();
|
||||
let record = accepted_record("invocation-a");
|
||||
InvocationReconciliationStore::new(&root.0)
|
||||
.insert(record.clone())
|
||||
.expect("insert record");
|
||||
|
||||
let loaded = InvocationReconciliationStore::new(&root.0)
|
||||
.get("invocation-a")
|
||||
.expect("load record")
|
||||
.expect("record exists");
|
||||
assert_eq!(loaded, record);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn running_transition_only_accepts_an_accepted_invocation() {
|
||||
let root = TestDirectory::new();
|
||||
let store = InvocationReconciliationStore::new(&root.0);
|
||||
store
|
||||
.insert(accepted_record("invocation-a"))
|
||||
.expect("insert");
|
||||
store.mark_running("invocation-a").expect("mark running");
|
||||
assert_eq!(
|
||||
store.get("invocation-a").unwrap().unwrap().status,
|
||||
InvocationStatus::Running
|
||||
);
|
||||
assert!(store.mark_running("invocation-a").is_err());
|
||||
store
|
||||
.mark_uncertain("invocation-a", "recovered")
|
||||
.expect("mark uncertain");
|
||||
assert!(store.mark_running("invocation-a").is_err());
|
||||
assert_eq!(
|
||||
store.get("invocation-a").unwrap().unwrap().status,
|
||||
InvocationStatus::Uncertain
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn uncertain_invocation_can_become_terminal_only_once() {
|
||||
let root = TestDirectory::new();
|
||||
let store = InvocationReconciliationStore::new(&root.0);
|
||||
store
|
||||
.insert(accepted_record("invocation-a"))
|
||||
.expect("insert");
|
||||
store
|
||||
.mark_uncertain("invocation-a", "runtime disconnected")
|
||||
.expect("mark uncertain");
|
||||
store
|
||||
.record_terminal(
|
||||
"invocation-a",
|
||||
InvocationOutcome::Completed { message_count: 2 },
|
||||
Some(serde_json::json!({"type": "completed"})),
|
||||
)
|
||||
.expect("record terminal");
|
||||
|
||||
let terminal = store.get("invocation-a").expect("load").expect("exists");
|
||||
assert_eq!(terminal.status, InvocationStatus::Terminal);
|
||||
assert_eq!(terminal.uncertainty_reason, None);
|
||||
assert_eq!(
|
||||
terminal.outcome,
|
||||
Some(InvocationOutcome::Completed { message_count: 2 })
|
||||
);
|
||||
assert!(store
|
||||
.record_terminal(
|
||||
"invocation-a",
|
||||
InvocationOutcome::Aborted { message_count: 0 },
|
||||
None,
|
||||
)
|
||||
.is_err());
|
||||
assert_eq!(store.get("invocation-a").expect("reload"), Some(terminal));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn recovery_marks_only_accepted_and_running_records_uncertain() {
|
||||
let root = TestDirectory::new();
|
||||
let store = InvocationReconciliationStore::new(&root.0);
|
||||
store
|
||||
.insert(accepted_record("accepted"))
|
||||
.expect("insert accepted");
|
||||
let mut running = accepted_record("running");
|
||||
running.status = InvocationStatus::Running;
|
||||
store.insert(running).expect("insert running");
|
||||
let mut terminal = accepted_record("terminal");
|
||||
terminal.status = InvocationStatus::Terminal;
|
||||
terminal.outcome = Some(InvocationOutcome::Aborted { message_count: 0 });
|
||||
store.insert(terminal.clone()).expect("insert terminal");
|
||||
|
||||
let recovered = store.reconcile_unfinished().expect("reconcile");
|
||||
assert_eq!(recovered.len(), 2);
|
||||
assert!(recovered
|
||||
.iter()
|
||||
.all(|record| record.status == InvocationStatus::Uncertain));
|
||||
assert_eq!(
|
||||
store.get("terminal").expect("load terminal"),
|
||||
Some(terminal)
|
||||
);
|
||||
assert!(store
|
||||
.reconcile_unfinished()
|
||||
.expect("repeat recovery")
|
||||
.is_empty());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn pending_terminal_result_survives_uncertainty_and_can_be_finalized() {
|
||||
let root = TestDirectory::new();
|
||||
let store = InvocationReconciliationStore::new(&root.0);
|
||||
store
|
||||
.insert(accepted_record("pending"))
|
||||
.expect("insert accepted record");
|
||||
store.mark_running("pending").expect("mark running");
|
||||
store
|
||||
.record_pending_terminal(
|
||||
"pending",
|
||||
InvocationOutcome::Completed { message_count: 3 },
|
||||
serde_json::json!({"eventType": "turn_ended"}),
|
||||
)
|
||||
.expect("record pending result");
|
||||
store
|
||||
.mark_uncertain("pending", "terminal scope store was unavailable")
|
||||
.expect("mark uncertain");
|
||||
|
||||
let pending = store.get("pending").unwrap().unwrap();
|
||||
assert_eq!(pending.status, InvocationStatus::Uncertain);
|
||||
assert_eq!(
|
||||
pending.outcome,
|
||||
Some(InvocationOutcome::Completed { message_count: 3 })
|
||||
);
|
||||
store
|
||||
.record_terminal_with_scope_revision(
|
||||
"pending",
|
||||
pending.outcome.unwrap(),
|
||||
pending.terminal_event,
|
||||
Some(1),
|
||||
)
|
||||
.expect("finalize after recovery");
|
||||
assert_eq!(
|
||||
store.get("pending").unwrap().unwrap().status,
|
||||
InvocationStatus::Terminal
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn concurrent_terminal_and_uncertain_transitions_preserve_terminal_result() {
|
||||
let root = TestDirectory::new();
|
||||
let store = InvocationReconciliationStore::new(&root.0);
|
||||
store
|
||||
.insert(accepted_record("concurrent"))
|
||||
.expect("insert accepted record");
|
||||
store.mark_running("concurrent").expect("mark running");
|
||||
|
||||
let barrier = std::sync::Arc::new(std::sync::Barrier::new(3));
|
||||
let terminal_store = store.clone();
|
||||
let terminal_barrier = std::sync::Arc::clone(&barrier);
|
||||
let terminal = std::thread::spawn(move || {
|
||||
terminal_barrier.wait();
|
||||
terminal_store.record_terminal(
|
||||
"concurrent",
|
||||
InvocationOutcome::Completed { message_count: 1 },
|
||||
None,
|
||||
)
|
||||
});
|
||||
let uncertain_store = store.clone();
|
||||
let uncertain_barrier = std::sync::Arc::clone(&barrier);
|
||||
let uncertain = std::thread::spawn(move || {
|
||||
uncertain_barrier.wait();
|
||||
uncertain_store.mark_uncertain("concurrent", "raced with terminal write")
|
||||
});
|
||||
barrier.wait();
|
||||
let _ = terminal.join().expect("terminal writer thread");
|
||||
let _ = uncertain.join().expect("uncertain writer thread");
|
||||
|
||||
let record = store.get("concurrent").unwrap().unwrap();
|
||||
assert_eq!(record.status, InvocationStatus::Terminal);
|
||||
assert_eq!(
|
||||
record.outcome,
|
||||
Some(InvocationOutcome::Completed { message_count: 1 })
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn concurrent_duplicate_inserts_have_only_one_winner() {
|
||||
let root = TestDirectory::new();
|
||||
let store = InvocationReconciliationStore::new(&root.0);
|
||||
let barrier = std::sync::Arc::new(std::sync::Barrier::new(3));
|
||||
let first_store = store.clone();
|
||||
let first_barrier = std::sync::Arc::clone(&barrier);
|
||||
let first = std::thread::spawn(move || {
|
||||
first_barrier.wait();
|
||||
first_store.insert(accepted_record("duplicate"))
|
||||
});
|
||||
let second_store = store.clone();
|
||||
let second_barrier = std::sync::Arc::clone(&barrier);
|
||||
let second = std::thread::spawn(move || {
|
||||
second_barrier.wait();
|
||||
second_store.insert(accepted_record("duplicate"))
|
||||
});
|
||||
barrier.wait();
|
||||
let first = first.join().expect("first insert thread");
|
||||
let second = second.join().expect("second insert thread");
|
||||
|
||||
assert_ne!(first.is_ok(), second.is_ok());
|
||||
assert_eq!(
|
||||
store.get("duplicate").unwrap().unwrap().invocation_id,
|
||||
"duplicate"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn malformed_ids_and_json_are_rejected() {
|
||||
let root = TestDirectory::new();
|
||||
let store = InvocationReconciliationStore::new(&root.0);
|
||||
assert!(store.get("../escape").is_err());
|
||||
assert!(store.insert(accepted_record("bad/id")).is_err());
|
||||
|
||||
fs::write(root.0.join("malformed.json"), "{").expect("write malformed record");
|
||||
assert!(store.get("malformed").is_err());
|
||||
assert!(store.list().is_err());
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,310 @@
|
||||
//! Durable mapping from SDK correlation keys to their active terminal session.
|
||||
|
||||
use std::{
|
||||
fs,
|
||||
path::PathBuf,
|
||||
sync::Mutex,
|
||||
time::{SystemTime, UNIX_EPOCH},
|
||||
};
|
||||
|
||||
use serde::{Deserialize, Serialize};
|
||||
|
||||
use crate::error::{SdkError, SdkResult};
|
||||
|
||||
use super::reconciliation::atomic_write_json;
|
||||
|
||||
const MAX_CORRELATION_KEY_BYTES: usize = 512;
|
||||
static TERMINAL_SCOPE_WRITE_LOCK: Mutex<()> = Mutex::new(());
|
||||
|
||||
/// Persisted terminal-session scope for one correlation key.
|
||||
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
|
||||
pub struct TerminalScopeRecord {
|
||||
pub correlation_key: String,
|
||||
pub session_id: String,
|
||||
pub revision: u64,
|
||||
pub updated_at_ms: u64,
|
||||
}
|
||||
|
||||
/// File-backed mapping of correlation keys to terminal sessions.
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct TerminalScopeStore {
|
||||
root: PathBuf,
|
||||
}
|
||||
|
||||
impl TerminalScopeStore {
|
||||
/// Creates a store handle without creating its root directory.
|
||||
#[must_use]
|
||||
pub fn new(root: impl Into<PathBuf>) -> Self {
|
||||
Self { root: root.into() }
|
||||
}
|
||||
|
||||
/// Sets the session associated with a correlation key.
|
||||
pub fn set(
|
||||
&self,
|
||||
correlation_key: impl Into<String>,
|
||||
session_id: impl Into<String>,
|
||||
) -> SdkResult<TerminalScopeRecord> {
|
||||
let correlation_key = correlation_key.into();
|
||||
validate_correlation_key(&correlation_key)?;
|
||||
let _guard = TERMINAL_SCOPE_WRITE_LOCK
|
||||
.lock()
|
||||
.unwrap_or_else(std::sync::PoisonError::into_inner);
|
||||
let previous = self.read_record(&correlation_key)?;
|
||||
let revision = previous
|
||||
.as_ref()
|
||||
.map_or(Some(1), |record| record.revision.checked_add(1))
|
||||
.ok_or_else(|| SdkError::internal("terminal scope revision is exhausted"))?;
|
||||
let previous_updated_at = previous.map_or(0, |record| record.updated_at_ms);
|
||||
let record = TerminalScopeRecord {
|
||||
correlation_key: correlation_key.clone(),
|
||||
session_id: session_id.into(),
|
||||
revision,
|
||||
updated_at_ms: current_timestamp_ms().max(previous_updated_at.saturating_add(1)),
|
||||
};
|
||||
atomic_write_json(&self.record_path(&correlation_key), &record)?;
|
||||
Ok(record)
|
||||
}
|
||||
|
||||
/// Rebuilds a mapping from durable invocation history without moving it backward.
|
||||
pub fn restore_if_newer(
|
||||
&self,
|
||||
correlation_key: &str,
|
||||
session_id: &str,
|
||||
revision: u64,
|
||||
) -> SdkResult<()> {
|
||||
validate_correlation_key(correlation_key)?;
|
||||
let _guard = TERMINAL_SCOPE_WRITE_LOCK
|
||||
.lock()
|
||||
.unwrap_or_else(std::sync::PoisonError::into_inner);
|
||||
let previous = self.read_record(correlation_key)?;
|
||||
match previous.as_ref() {
|
||||
Some(record) if record.revision > revision => return Ok(()),
|
||||
Some(record) if record.revision == revision && record.session_id == session_id => {
|
||||
return Ok(())
|
||||
}
|
||||
Some(record) if record.revision == revision => {
|
||||
return Err(SdkError::internal(format!(
|
||||
"terminal scope revision {revision} conflicts for key '{correlation_key}'"
|
||||
)));
|
||||
}
|
||||
_ => {}
|
||||
}
|
||||
let record = TerminalScopeRecord {
|
||||
correlation_key: correlation_key.to_string(),
|
||||
session_id: session_id.to_string(),
|
||||
revision,
|
||||
updated_at_ms: current_timestamp_ms().max(
|
||||
previous
|
||||
.map_or(0, |record| record.updated_at_ms)
|
||||
.saturating_add(1),
|
||||
),
|
||||
};
|
||||
atomic_write_json(&self.record_path(correlation_key), &record)
|
||||
}
|
||||
|
||||
/// Loads the session associated with a correlation key, if present.
|
||||
pub fn get(&self, correlation_key: &str) -> SdkResult<Option<TerminalScopeRecord>> {
|
||||
validate_correlation_key(correlation_key)?;
|
||||
self.read_record(correlation_key)
|
||||
}
|
||||
|
||||
fn read_record(&self, correlation_key: &str) -> SdkResult<Option<TerminalScopeRecord>> {
|
||||
let path = self.record_path(correlation_key);
|
||||
let bytes = match fs::read(&path) {
|
||||
Ok(bytes) => bytes,
|
||||
Err(error) if error.kind() == std::io::ErrorKind::NotFound => return Ok(None),
|
||||
Err(error) => {
|
||||
return Err(SdkError::internal(format!(
|
||||
"read terminal scope '{}': {error}",
|
||||
path.display()
|
||||
)));
|
||||
}
|
||||
};
|
||||
let record: TerminalScopeRecord = serde_json::from_slice(&bytes).map_err(|error| {
|
||||
SdkError::internal(format!(
|
||||
"parse terminal scope '{}': {error}",
|
||||
path.display()
|
||||
))
|
||||
})?;
|
||||
if record.correlation_key != correlation_key {
|
||||
return Err(SdkError::internal(format!(
|
||||
"terminal scope key does not match requested key for '{}'",
|
||||
path.display()
|
||||
)));
|
||||
}
|
||||
Ok(Some(record))
|
||||
}
|
||||
|
||||
fn record_path(&self, correlation_key: &str) -> PathBuf {
|
||||
const MAX_COMPONENT_BYTES: usize = 180;
|
||||
let encoded = encode_key_as_hex(correlation_key);
|
||||
let mut path = self.root.clone();
|
||||
let mut remaining = encoded.as_str();
|
||||
while remaining.len() > MAX_COMPONENT_BYTES {
|
||||
let (component, tail) = remaining.split_at(MAX_COMPONENT_BYTES);
|
||||
path.push(component);
|
||||
remaining = tail;
|
||||
}
|
||||
path.push(format!("{remaining}.json"));
|
||||
path
|
||||
}
|
||||
}
|
||||
|
||||
fn validate_correlation_key(correlation_key: &str) -> SdkResult<()> {
|
||||
if correlation_key.len() <= MAX_CORRELATION_KEY_BYTES {
|
||||
Ok(())
|
||||
} else {
|
||||
Err(SdkError::invalid_input("correlation key exceeds 512 bytes"))
|
||||
}
|
||||
}
|
||||
|
||||
fn encode_key_as_hex(key: &str) -> String {
|
||||
const HEX_DIGITS: &[u8; 16] = b"0123456789abcdef";
|
||||
let mut encoded = String::with_capacity(key.len() * 2);
|
||||
for byte in key.as_bytes() {
|
||||
encoded.push(char::from(HEX_DIGITS[usize::from(*byte >> 4)]));
|
||||
encoded.push(char::from(HEX_DIGITS[usize::from(*byte & 0x0f)]));
|
||||
}
|
||||
encoded
|
||||
}
|
||||
|
||||
fn current_timestamp_ms() -> u64 {
|
||||
SystemTime::now()
|
||||
.duration_since(UNIX_EPOCH)
|
||||
.unwrap_or_default()
|
||||
.as_millis()
|
||||
.min(u128::from(u64::MAX)) as u64
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use std::fs;
|
||||
use uuid::Uuid;
|
||||
|
||||
struct TestDirectory(PathBuf);
|
||||
|
||||
impl TestDirectory {
|
||||
fn new() -> Self {
|
||||
let path = std::env::temp_dir().join(format!("sdk-terminal-scope-{}", Uuid::new_v4()));
|
||||
fs::create_dir(&path).expect("create test directory");
|
||||
Self(path)
|
||||
}
|
||||
}
|
||||
|
||||
impl Drop for TestDirectory {
|
||||
fn drop(&mut self) {
|
||||
let _ = fs::remove_dir_all(&self.0);
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn terminal_scope_survives_store_restart() {
|
||||
let root = TestDirectory::new();
|
||||
TerminalScopeStore::new(&root.0)
|
||||
.set("request-1", "session-1")
|
||||
.expect("set terminal scope");
|
||||
|
||||
let loaded = TerminalScopeStore::new(&root.0)
|
||||
.get("request-1")
|
||||
.expect("load terminal scope")
|
||||
.expect("scope exists");
|
||||
assert_eq!(loaded.correlation_key, "request-1");
|
||||
assert_eq!(loaded.session_id, "session-1");
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn scope_revisions_increase_and_recovery_never_moves_backward() {
|
||||
let root = TestDirectory::new();
|
||||
let store = TerminalScopeStore::new(&root.0);
|
||||
let first = store.set("request-1", "session-1").unwrap();
|
||||
let second = store.set("request-1", "session-2").unwrap();
|
||||
assert_eq!(second.revision, first.revision + 1);
|
||||
assert!(second.updated_at_ms > first.updated_at_ms);
|
||||
|
||||
store
|
||||
.restore_if_newer("request-1", "session-1", first.revision)
|
||||
.expect("stale scope restoration should be ignored");
|
||||
assert_eq!(
|
||||
store.get("request-1").unwrap().unwrap().session_id,
|
||||
"session-2"
|
||||
);
|
||||
assert!(store
|
||||
.restore_if_newer("request-1", "other-session", second.revision)
|
||||
.is_err());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn encoded_keys_are_isolated_and_cannot_traverse_paths() {
|
||||
let root = TestDirectory::new();
|
||||
let store = TerminalScopeStore::new(&root.0);
|
||||
store.set("same", "session-a").expect("set first scope");
|
||||
store
|
||||
.set("../same", "session-b")
|
||||
.expect("set path-like key");
|
||||
store
|
||||
.set("../sämé", "session-c")
|
||||
.expect("set Unicode path-like key");
|
||||
|
||||
assert_eq!(
|
||||
store
|
||||
.get("same")
|
||||
.expect("get first")
|
||||
.expect("exists")
|
||||
.session_id,
|
||||
"session-a"
|
||||
);
|
||||
assert_eq!(
|
||||
store
|
||||
.get("../same")
|
||||
.expect("get path-like key")
|
||||
.expect("exists")
|
||||
.session_id,
|
||||
"session-b"
|
||||
);
|
||||
assert_eq!(
|
||||
store
|
||||
.get("../sämé")
|
||||
.expect("get Unicode path-like key")
|
||||
.expect("exists")
|
||||
.session_id,
|
||||
"session-c"
|
||||
);
|
||||
assert_eq!(fs::read_dir(&root.0).expect("read root").count(), 3);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn malformed_terminal_scope_json_is_rejected() {
|
||||
let root = TestDirectory::new();
|
||||
let store = TerminalScopeStore::new(&root.0);
|
||||
let path = root
|
||||
.0
|
||||
.join(format!("{}.json", encode_key_as_hex("request")));
|
||||
fs::write(path, "{").expect("write malformed scope");
|
||||
assert!(store.get("request").is_err());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn long_hex_keys_use_safe_path_components() {
|
||||
let root = TestDirectory::new();
|
||||
let store = TerminalScopeStore::new(&root.0);
|
||||
let key = "k".repeat(512);
|
||||
store
|
||||
.set(&key, "session-long")
|
||||
.expect("long key should persist");
|
||||
assert_eq!(
|
||||
store.get(&key).expect("get long key").unwrap().session_id,
|
||||
"session-long"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn oversized_correlation_keys_are_rejected() {
|
||||
let root = TestDirectory::new();
|
||||
let store = TerminalScopeStore::new(&root.0);
|
||||
let key = "k".repeat(513);
|
||||
assert!(store.set(&key, "session-long").is_err());
|
||||
assert!(store.get(&key).is_err());
|
||||
assert!(!root.0.exists() || fs::read_dir(&root.0).unwrap().next().is_none());
|
||||
}
|
||||
}
|
||||
File diff suppressed because it is too large
Load Diff
Reference in New Issue
Block a user