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
4 changes: 4 additions & 0 deletions docs/remote-protocol/APP-UI.md
Original file line number Diff line number Diff line change
Expand Up @@ -145,6 +145,10 @@ observed terminal time and outcome can still be supplied.
logical turn metadata by `turn_id` and durable messages by
`(history_revision, message_id)`.
2. An attached snapshot replaces session state at `snapshot.sequence`.
Keep its optional `snapshot_id` in the reconnect cursor as well as the
applied sequence, gateway ID, instance ID and registration epoch. Send
`snapshot_id` in `attach_session.resume` to resume at the current watermark;
refresh it on every replacement snapshot, and retain it across later events.
Restore active clocks from `turn.timing`, terminal footers from `last_turn`
and `transcript[].turn`, and thinking labels from each assistant segment.
3. Apply subsequent numbered events exactly once. Replace timing values;
Expand Down
25 changes: 24 additions & 1 deletion docs/remote-protocol/README.md
Original file line number Diff line number Diff line change
Expand Up @@ -294,7 +294,7 @@ attaches with that cursor:
{
"protocol_version": 1, "request_id": "req-0003",
"session_id": "…", "registration_epoch": 3,
"operation": { "type": "attach_session", "resume": { "gateway_id": "…", "instance_id": "…", "last_sequence": 412 } }
"operation": { "type": "attach_session", "resume": { "gateway_id": "…", "instance_id": "…", "last_sequence": 412, "snapshot_id": "…" } }
}
```

Expand Down Expand Up @@ -326,6 +326,20 @@ sequence alone, which also refuses every cursor from before a restart (as
`sequence_gap` rather than `gateway_restarted`). A cursor for another
`registration_epoch` is `stale_session`, as for any request.

Store `snapshot.snapshot_id` alongside the applied sequence and send it as
`resume.snapshot_id`. This optional opaque identifier lets a cursor exactly
at the current snapshot watermark resume, including sequence zero with no
missed events. Snapshots can replace state without advancing the sequence:
a replaced snapshot has a new identifier, so its old cursor gets a fresh
`attached` response instead. Keep the identifier while applying later events;
replace it whenever a snapshot replaces the local state. Older clients without
the identifier still work, but get a fresh snapshot at this boundary.

New transcript messages use RFC3339 `timestamp` values with a timezone.
Legacy time-only values have no known date and are omitted from the remote
projection. An untitled session with no user prompt has an empty title;
the app can choose its own display placeholder.

An `attached` without a requested `resume` has no `resync` field.

Attaching again on a connection that is already attached to the session
Expand Down Expand Up @@ -447,6 +461,15 @@ discover them:
custom answer is allowed; the terminal accepts a non-empty `custom` answer
for every question.
- Approval actions have no ID; a batch is approved or denied as a whole.
- Mutation receipts are bounded by count (1024 per registration), with no age
eviction. At capacity, new mutations return `receipt_capacity`; known IDs
remain queryable. Stop and re-share the session to create a new receipt store.
- Tool calls expose bounded summary/input/output text, without a separate
structured command field. A tool-specific command copy button needs an
additive structured payload; do not reconstruct commands from summaries.
- On Unix, a fallback `/tmp/rustcode-<uid>` socket directory can remain after
gateway shutdown. Socket files are removed; the private directory is retained
to avoid racing another process using it. Clients never need that path.
- `resync_required` with reason `gateway_restarted` is never sent as an
event: a restart closes the connection, and the reason reaches the client
in `attached.resync`.
Expand Down
7 changes: 7 additions & 0 deletions docs/remote-protocol/v1/frame.schema.json
Original file line number Diff line number Diff line change
Expand Up @@ -1329,6 +1329,13 @@
"session": {
"$ref": "#/$defs/RemoteSessionInfo"
},
"snapshot_id": {
"description": "Opaque gateway identity for this snapshot cut, including replacements\nat the same sequence. Store alongside the reconnect cursor.",
"type": [
"string",
"null"
]
},
"subagents": {
"items": {
"$ref": "#/$defs/RemoteSubagent"
Expand Down
1 change: 1 addition & 0 deletions docs/remote-protocol/v1/golden/frames/event_snapshot.json
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,7 @@
},
"health": "live"
},
"snapshot_id": "snapshot-example",
"sequence": 424,
"generation": 1,
"last_turn": {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@
},
"health": "live"
},
"snapshot_id": "snapshot-example",
"sequence": 412,
"generation": 1,
"turn": {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,7 @@
},
"health": "live"
},
"snapshot_id": "snapshot-example",
"sequence": 412,
"generation": 1,
"turn": {
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,7 @@
"type": "attach_session",
"resume": {
"gateway_id": "gw-8e1f4b7a",
"snapshot_id": "snapshot-example",
"instance_id": "inst-41c07d92",
"last_sequence": 412
}
Expand Down
7 changes: 7 additions & 0 deletions docs/remote-protocol/v1/request.schema.json
Original file line number Diff line number Diff line change
Expand Up @@ -308,6 +308,13 @@
"format": "uint64",
"minimum": 0,
"type": "integer"
},
"snapshot_id": {
"description": "Identity of the last applied snapshot. Required to resume exactly at\nits watermark, where a newer snapshot can have the same sequence.",
"type": [
"string",
"null"
]
}
},
"required": [
Expand Down
4 changes: 2 additions & 2 deletions rustcode/core/rustcode-core/src/lib.rs
Original file line number Diff line number Diff line change
Expand Up @@ -15,7 +15,7 @@ pub mod status;
pub mod update;

fn current_timestamp() -> String {
chrono::Local::now().format("%H:%M").to_string()
chrono::Local::now().to_rfc3339()
}

/// Usage reported by a provider for one completion request.
Expand Down Expand Up @@ -361,7 +361,7 @@ pub struct ChatMessage {
pub turn: Option<TurnTiming>,
#[serde(skip_serializing_if = "Option::is_none")]
pub token_usage: Option<TokenUsage>,
#[serde(default = "current_timestamp")]
#[serde(default)]
pub timestamp: String,
#[serde(skip_serializing_if = "Option::is_none")]
pub response_time_ms: Option<u64>,
Expand Down
4 changes: 3 additions & 1 deletion rustcode/engine/src/app/subagent_context.rs
Original file line number Diff line number Diff line change
Expand Up @@ -160,7 +160,9 @@ mod tests {
ContextInheritance::Minimal,
Some("evidence"),
);
assert_eq!(minimal, vec![ChatMessage::new("user", "child")]);
assert_eq!(minimal.len(), 1);
assert_eq!(minimal[0].role, "user");
assert_eq!(minimal[0].content, "child");
let evidence =
inherit_context(&parent, "child", ContextInheritance::Evidence, Some("fact"));
assert_eq!(evidence.len(), 2);
Expand Down
6 changes: 2 additions & 4 deletions rustcode/engine/src/app/subagent_controller.rs
Original file line number Diff line number Diff line change
Expand Up @@ -909,10 +909,8 @@ mod tests {
assert_eq!(contexts[1].parent_id, Some(parent));
assert_eq!(contexts[1].status, SubAgentStatus::Running);
assert!(contexts[1].active_turn);
assert_eq!(
contexts[1].history[0],
ChatMessage::new("user", "inspect the child")
);
assert_eq!(contexts[1].history[0].role, "user");
assert_eq!(contexts[1].history[0].content, "inspect the child");
}

#[test]
Expand Down
2 changes: 2 additions & 0 deletions rustcode/engine/src/remote/contract.rs
Original file line number Diff line number Diff line change
Expand Up @@ -244,6 +244,7 @@ pub fn golden_requests() -> Vec<(&'static str, RemoteRequest)> {
RemoteOperation::AttachSession {
resume: Some(ResumeCursor {
gateway_id: GATEWAY.to_owned(),
snapshot_id: Some("snapshot-example".to_owned()),
instance_id: Some(INSTANCE.to_owned()),
last_sequence: 412,
}),
Expand Down Expand Up @@ -536,6 +537,7 @@ fn golden_previous_timing() -> RemoteTurnTiming {

fn golden_snapshot() -> RemoteSnapshot {
RemoteSnapshot {
snapshot_id: Some("snapshot-example".to_owned()),
session: golden_session_info(),
sequence: 412,
generation: 1,
Expand Down
52 changes: 47 additions & 5 deletions rustcode/engine/src/remote/projection.rs
Original file line number Diff line number Diff line change
Expand Up @@ -121,10 +121,17 @@ pub fn project_session_info(state: &AppState, context: &ProjectionContext) -> Re
RemoteSessionInfo {
session_id: state.active_session_id.clone(),
registration_epoch: context.registration_epoch,
title: context
.title
.clone()
.unwrap_or_else(|| crate::config::session_title(&state.history)),
title: context.title.clone().unwrap_or_else(|| {
if state
.history
.iter()
.any(|message| message.role == "user" && !message.content.starts_with('/'))
{
crate::config::session_title(&state.history)
} else {
String::new()
}
}),
workspace: state
.task_working_directory
.clone()
Expand Down Expand Up @@ -300,7 +307,9 @@ fn project_messages(
success: record.success,
pending: record.pending,
}),
timestamp: Some(message.timestamp.clone()).filter(|timestamp| !timestamp.is_empty()),
timestamp: chrono::DateTime::parse_from_rfc3339(&message.timestamp)
.ok()
.map(|_| message.timestamp.clone()),
response_time_ms: message.response_time_ms,
thought_time_ms: message.thought_time_ms,
completed_at: message.completed_at.clone(),
Expand Down Expand Up @@ -435,6 +444,7 @@ pub fn project_snapshot(
.collect();
let mut snapshot = RemoteSnapshot {
session: project_session_info(state, context),
snapshot_id: None,
sequence: context.sequence,
generation: context.generation,
turn,
Expand Down Expand Up @@ -685,6 +695,38 @@ pub fn project_event(

#[cfg(test)]
mod tests {
#[test]
fn remote_client_followups_use_machine_timestamps_and_empty_untitled_sessions() {
let mut state = state_with_history(0, 0);
let mut context = context();
context.title = None;
assert!(project_session_info(&state, &context).title.is_empty());
state
.history
.push(crate::app::ChatMessage::new("user", "fresh prompt"));
let page = project_history_page(&state, &ProjectionLimits::default(), None, 10).unwrap();
assert!(
chrono::DateTime::parse_from_rfc3339(page.messages[0].timestamp.as_deref().unwrap())
.is_ok()
);
state.history.as_mut_vec()[0].timestamp = "18:42".into();
let page = project_history_page(&state, &ProjectionLimits::default(), None, 10).unwrap();
assert!(
page.messages[0].timestamp.is_none(),
"legacy time-only values have no known date"
);
state.history.push(
serde_json::from_value(serde_json::json!({
"role": "assistant", "content": "Legacy answer without a timestamp"
}))
.unwrap(),
);
let page = project_history_page(&state, &ProjectionLimits::default(), None, 10).unwrap();
assert!(state.history[1].timestamp.is_empty());
assert!(page.messages[1].timestamp.is_none());
context.title = Some("My title".into());
assert_eq!(project_session_info(&state, &context).title, "My title");
}
use super::*;
use crate::app::{AppStatus, ChatMessage, ToolConfirmation};

Expand Down
8 changes: 8 additions & 0 deletions rustcode/engine/src/remote/protocol.rs
Original file line number Diff line number Diff line change
Expand Up @@ -160,6 +160,10 @@ impl RemoteOperation {
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
pub struct ResumeCursor {
pub gateway_id: String,
/// Identity of the last applied snapshot. Required to resume exactly at
/// its watermark, where a newer snapshot can have the same sequence.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub snapshot_id: Option<String>,
/// The `instance_id` the cursor was obtained under, as the handshake and
/// the `attached`/`resumed` results report it. A cursor from another
/// gateway instance is never replayed. When absent the gateway decides
Expand Down Expand Up @@ -565,6 +569,10 @@ pub struct RemoteContentChunk {
#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize, JsonSchema)]
pub struct RemoteSnapshot {
pub session: RemoteSessionInfo,
/// Opaque gateway identity for this snapshot cut, including replacements
/// at the same sequence. Store alongside the reconnect cursor.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub snapshot_id: Option<String>,
pub sequence: u64,
pub generation: u64,
/// Present while a prompt is being run. Absent in the short gap between
Expand Down
48 changes: 48 additions & 0 deletions rustcode/engine/src/remote_gateway/e2e_tests.rs
Original file line number Diff line number Diff line change
Expand Up @@ -490,6 +490,54 @@ impl View {
}
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn gateway_resumes_the_current_snapshot_and_exposes_machine_readable_history() {
let host = Host::start().await;
let connector = host.connector();
let mut terminal = Terminal::share(&connector, "session-client-followups").await;
terminal.registered().await;
let (mut phone, paired) = Phone::pair(&host, "Cursor phone").await;
let attached = phone
.request(
Some(&terminal.registration),
json!({"type": "attach_session"}),
)
.await;
let snapshot = &attached["result"]["snapshot"];
assert_eq!(snapshot["session"]["title"], "");
assert!(snapshot["snapshot_id"].is_string());
let cursor = json!({"gateway_id": paired["gateway_id"], "instance_id": paired["instance_id"],
"last_sequence": snapshot["sequence"], "snapshot_id": snapshot["snapshot_id"]});
drop(phone);
let (mut phone, _) = Phone::authenticate(&host, &paired).await;
let resumed = phone
.request(
Some(&terminal.registration),
json!({"type": "attach_session", "resume": cursor}),
)
.await;
assert_eq!(resumed["result"]["type"], "resumed", "{resumed}");
terminal
.state
.lock()
.await
.history
.push(crate::app::ChatMessage::new("user", "Fresh timestamp"));
let history = phone
.request(
Some(&terminal.registration),
json!({"type": "get_history", "limit": 10}),
)
.await;
let timestamp = history["result"]["messages"][0]["timestamp"]
.as_str()
.unwrap();
assert!(
chrono::DateTime::parse_from_rfc3339(timestamp).is_ok(),
"{timestamp}"
);
}

#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
async fn repeated_remote_off_preserves_the_reason_with_gateway_status_in_flight() {
let host = Host::start().await;
Expand Down
Loading
Loading