Compare commits
2
Commits
4add4fd176
...
6d12388063
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
6d12388063 | ||
|
|
465645cefb |
No files matched your search
@@ -286,8 +286,12 @@ first if a remote spawn ever mangles an argument.
|
||||
cross-session message to a real stream-json session: no `user` record, and
|
||||
nothing in the partial-message stream -- the whole of it is an `origin`
|
||||
object on the `result`, the same shape the session file records, which is
|
||||
why `import::peer_message` reads both. So the note is drawn *after* the
|
||||
reply it caused; that is the wire, not a bug. See PLAN.md.
|
||||
why `import::peer_message` reads both. So it is *recorded* after the reply
|
||||
it caused, and cannot be recorded anywhere else in an append-only log --
|
||||
which is why the event carries `turnStart`, the seq of the status that
|
||||
opened its turn, and the phone draws the note at that seq instead of where
|
||||
it arrived. Exercise it with the echo driver's `/peer-turn`; plain `/peer`
|
||||
is the in-place shape an import replays. See PLAN.md.
|
||||
- **A queued message can be tapped to take it back**, which is
|
||||
`POST /sessions/{id}/unqueue` and a `messageDropped` event -- see PLAN.md's
|
||||
"Taking a queued message back". On a **Claude** session it always refuses,
|
||||
|
||||
@@ -300,13 +300,30 @@ both, and there is one function for one wire format. Only peer-caused turns
|
||||
carry it: four ordinary results on a real session's stdout had no `origin`
|
||||
between them.
|
||||
|
||||
**The cost is the position.** The note lands after the reply it caused rather
|
||||
than above it, because at no earlier point in the turn does the CLI say why
|
||||
the turn started. The alternative is a second reader tailing the CLI's own
|
||||
session file for the one record stdout does not carry — two sources of truth
|
||||
for one conversation and a poll per live session — and it was rejected on
|
||||
that. If the CLI ever announces the injection at the point it happens, this
|
||||
moves to that record and the ordering comes right with it.
|
||||
**The cost was the position, and it is paid on the wire rather than on
|
||||
screen** (2026-09-01). The event cannot be *recorded* in place: at no earlier
|
||||
point in the turn does the CLI say why the turn started, and the transcript
|
||||
is append-only, so by the time anyone knows, everything the message caused
|
||||
has already been written above it. Reading it out of the CLI's own session
|
||||
file instead — a second reader tailing the one record stdout does not carry —
|
||||
was rejected then and stays rejected: two sources of truth for one
|
||||
conversation and a poll per live session.
|
||||
|
||||
So the event carries **where it belongs** instead. `PeerMessage` has a
|
||||
`turnStart`: the seq of the `Status` that opened the turn it started, stamped
|
||||
by the pump, which is the only thing that knows a seq and the only thing that
|
||||
sees every driver's turns. The phone gives the note that seq, so it sorts
|
||||
into the transcript above the turn rather than being drawn out of order at
|
||||
the end. That seq belongs to a status change, and a status draws no row, so
|
||||
there is nothing for the note to collide with and the list stays sorted —
|
||||
which is what the scroll anchor and paging depend on.
|
||||
|
||||
`turnStart` is absent where there is nothing to correct: a message replayed
|
||||
out of a session file by `import` is already in the right place, and one that
|
||||
opened no turn has no turn to sit above. Both are drawn where they arrive.
|
||||
The echo driver models both shapes — `/peer` for the in-place one, and
|
||||
`/peer-turn` for the live one, which reveals the note only after a reply and
|
||||
a run of tool calls.
|
||||
|
||||
### Taking a queued message back (decided 2026-08-31)
|
||||
|
||||
|
||||
@@ -99,7 +99,19 @@ sealed class SessionEvent {
|
||||
* claim they had. It is also the explanation for a session that starts working on something
|
||||
* this device never asked for.
|
||||
*/
|
||||
data class PeerMessage(val from: String, val text: String) : SessionEvent()
|
||||
data class PeerMessage(
|
||||
val from: String,
|
||||
val text: String,
|
||||
/**
|
||||
* Where the turn this started begins, when the server could say.
|
||||
*
|
||||
* The live Claude Code path only learns a turn was somebody else's when the turn ends, so
|
||||
* the event arrives below everything it caused; this is what puts it back above it. Null
|
||||
* for a message read out of a session file, which is already in the right place, and for
|
||||
* one that started no turn. See the server's `Event::PeerMessage`.
|
||||
*/
|
||||
val turnStart: Long? = null,
|
||||
) : SessionEvent()
|
||||
|
||||
/**
|
||||
* A command the session was asked to run on itself and cannot run yet.
|
||||
@@ -237,7 +249,11 @@ fun parseSeqEvent(json: String): SeqEvent {
|
||||
},
|
||||
)
|
||||
"peerMessage" ->
|
||||
SessionEvent.PeerMessage(body.getString("from"), body.getString("text"))
|
||||
SessionEvent.PeerMessage(
|
||||
body.getString("from"),
|
||||
body.getString("text"),
|
||||
if (body.has("turnStart")) body.getLong("turnStart") else null,
|
||||
)
|
||||
"commandQueued" ->
|
||||
SessionEvent.CommandQueued(body.getString("id"), body.getString("text"))
|
||||
"commandSent" -> SessionEvent.CommandSent(body.getString("id"), body.getString("text"))
|
||||
|
||||
@@ -268,6 +268,56 @@ private fun adoptRun(
|
||||
tail.map { (it as TranscriptItem.ToolRun).copy(runId = joining) }
|
||||
}
|
||||
|
||||
/**
|
||||
* A peer message goes above the turn it started, not where it happened to arrive.
|
||||
*
|
||||
* The live Claude Code path cannot record it in place: the CLI says nothing about a peer message
|
||||
* until the turn's `result`, so the event lands below the whole reply it caused -- the answer
|
||||
* printed above the question. The server stamps it with where that turn began
|
||||
* ([SessionEvent.PeerMessage.turnStart]) and the note takes that seq, so it sorts into the list
|
||||
* where it belongs rather than being drawn out of order at the end.
|
||||
*
|
||||
* Taking the turn's opening seq as its own is also what keeps the list sorted, which anchors and
|
||||
* paging both depend on. That seq belongs to a status change, and a status draws no row, so there
|
||||
* is nothing for it to collide with.
|
||||
*
|
||||
* Without a stamp -- a message replayed out of a session file, which is already in the right place
|
||||
* -- it stays where it arrived.
|
||||
*/
|
||||
private fun placePeerNote(
|
||||
items: List<TranscriptItem>,
|
||||
seq: Long,
|
||||
event: SessionEvent.PeerMessage,
|
||||
): List<TranscriptItem> {
|
||||
val at = event.turnStart ?: return items + TranscriptItem.PeerNote(seq, event.from, event.text)
|
||||
val note = TranscriptItem.PeerNote(at, event.from, event.text)
|
||||
val index = items.indexOfFirst { it.seq > at }
|
||||
if (index < 0) return items + note
|
||||
val behind = (items.getOrNull(index - 1) as? TranscriptItem.ToolRun)?.runId
|
||||
return items.subList(0, index) + note + splitRun(items.subList(index, items.size), behind)
|
||||
}
|
||||
|
||||
/**
|
||||
* The calls the note now sits in front of, renamed if they were sharing a run with the calls behind
|
||||
* it.
|
||||
*
|
||||
* A run is named from what a call landed next to (see [runIdFor]), and nothing there knows about
|
||||
* turns -- so a turn opening with a tool call, straight after one that ended with one, folds them
|
||||
* into a single run. Left alone, [groupToolRuns] would flush at the note and hand both halves the
|
||||
* same name: two rows with one key, which a keyed list cannot draw at all.
|
||||
*
|
||||
* The later half is the one renamed, which is the opposite of a page join ([adoptRun]) and right
|
||||
* for the opposite reason. There the two halves were always one run and the newer was already on
|
||||
* screen; here they were never one turn's work, and both halves change appearance at the same
|
||||
* moment the note appears between them.
|
||||
*/
|
||||
private fun splitRun(tail: List<TranscriptItem>, behind: String?): List<TranscriptItem> {
|
||||
val first = tail.firstOrNull() as? TranscriptItem.ToolRun ?: return tail
|
||||
if (behind == null || first.runId != behind) return tail
|
||||
val run = tail.takeWhile { it is TranscriptItem.ToolRun && it.runId == behind }
|
||||
return run.map { (it as TranscriptItem.ToolRun).copy(runId = first.id) } + tail.drop(run.size)
|
||||
}
|
||||
|
||||
fun foldEvent(items: List<TranscriptItem>, entry: SeqEvent): List<TranscriptItem> =
|
||||
when (val event = entry.event) {
|
||||
is SessionEvent.UserMessage ->
|
||||
@@ -361,8 +411,7 @@ fun foldEvent(items: List<TranscriptItem>, entry: SeqEvent): List<TranscriptItem
|
||||
else -> it
|
||||
}
|
||||
}
|
||||
is SessionEvent.PeerMessage ->
|
||||
items + TranscriptItem.PeerNote(entry.seq, event.from, event.text)
|
||||
is SessionEvent.PeerMessage -> placePeerNote(items, entry.seq, event)
|
||||
is SessionEvent.CommandSent -> items + TranscriptItem.CommandRow(entry.seq, event.text)
|
||||
// Screen-level state, not transcript rows -- see SessionScreen.
|
||||
is SessionEvent.CommandQueued -> items
|
||||
|
||||
@@ -1147,6 +1147,9 @@ mod tests {
|
||||
Event::PeerMessage {
|
||||
from: "ai-app-2-fb".to_string(),
|
||||
text: "Reply with just the word ACK.".to_string(),
|
||||
// Stamped by the pump, which is the only place that
|
||||
// knows what seq the turn started at.
|
||||
turn_start: None,
|
||||
},
|
||||
Event::UsageDelta {
|
||||
tokens: 7,
|
||||
|
||||
@@ -225,6 +225,24 @@ pub enum Event {
|
||||
/// recognises it by -- the socket path it came from is not.
|
||||
from: String,
|
||||
text: String,
|
||||
/// The seq of the `Status::Running` that opened the turn this
|
||||
/// message started, so a reader can draw it above that turn.
|
||||
///
|
||||
/// It exists because the live Claude Code path cannot record the
|
||||
/// message where it belongs. The CLI says nothing about a peer
|
||||
/// message until the turn's `result` -- see
|
||||
/// `claude::translate` -- so the event is appended after
|
||||
/// everything it caused, and an append-only transcript cannot go
|
||||
/// back and insert it. Carrying the position instead keeps one
|
||||
/// order on the wire and one order on screen without a second
|
||||
/// source for either.
|
||||
///
|
||||
/// Filled in by the pump, which is the only place that knows a
|
||||
/// seq, and only where a turn was open: `None` for a message read
|
||||
/// out of a session file by `import`, which already has it in the
|
||||
/// right place, and for one that started no turn.
|
||||
#[serde(default, skip_serializing_if = "Option::is_none")]
|
||||
turn_start: Option<u64>,
|
||||
},
|
||||
/// The manager's record of a question being answered, so a rendered
|
||||
/// question card resolves on every device, not just the one that
|
||||
|
||||
@@ -278,6 +278,42 @@ impl EchoDriver {
|
||||
// asked for it is still announced -- every driver owes exactly one
|
||||
// `MessageTaken` per message, and a command that quietly vanishes
|
||||
// from the transcript is the one thing echo must not model.
|
||||
// The live Claude Code shape, which is the one the ordering has to
|
||||
// survive: the CLI says nothing about a peer message until the
|
||||
// turn's `result`, so the event arrives below the whole reply it
|
||||
// caused and the phone has to put it back. Checked before `/peer`,
|
||||
// which would otherwise take the rest of this word as the body.
|
||||
if let Some(rest) = text.strip_prefix("/peer-turn") {
|
||||
if announce {
|
||||
self.emit(Event::MessageTaken {
|
||||
id: None,
|
||||
text: text.clone(),
|
||||
images: images.clone(),
|
||||
});
|
||||
}
|
||||
self.emit(Event::Status {
|
||||
state: SessionStatus::Running,
|
||||
});
|
||||
self.some_calls("peer");
|
||||
self.emit(Event::AssistantText {
|
||||
delta: "Pulled, and AGENTS.md is up to date here now.".to_string(),
|
||||
});
|
||||
self.emit(Event::PeerMessage {
|
||||
from: "dev-updater-f5".to_string(),
|
||||
text: if rest.trim().is_empty() {
|
||||
"Pull before you touch AGENTS.md.".to_string()
|
||||
} else {
|
||||
rest.trim().to_string()
|
||||
},
|
||||
// Stamped by the manager, exactly as a real one is.
|
||||
turn_start: None,
|
||||
});
|
||||
self.emit(Event::Status {
|
||||
state: SessionStatus::Idle,
|
||||
});
|
||||
return;
|
||||
}
|
||||
|
||||
if let Some(rest) = text.strip_prefix("/peer") {
|
||||
if announce {
|
||||
self.emit(Event::MessageTaken {
|
||||
@@ -296,6 +332,7 @@ impl EchoDriver {
|
||||
} else {
|
||||
rest.trim().to_string()
|
||||
},
|
||||
turn_start: None,
|
||||
});
|
||||
return;
|
||||
}
|
||||
@@ -710,6 +747,7 @@ async fn write_beat(sink: &EventSink, session_dir: &Path, beat: usize) {
|
||||
send(Event::PeerMessage {
|
||||
from: format!("beat-{beat}-peer"),
|
||||
text: format!("Message {beat} from another session, for the row it becomes."),
|
||||
turn_start: None,
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
@@ -550,6 +550,8 @@ pub(in crate::session) fn peer_message(record: &Value) -> Option<Event> {
|
||||
.unwrap_or("another session")
|
||||
.to_string(),
|
||||
text: origin.get("body").and_then(Value::as_str)?.to_string(),
|
||||
// The session file has it in the right place already.
|
||||
turn_start: None,
|
||||
})
|
||||
}
|
||||
|
||||
@@ -1151,6 +1153,7 @@ mod tests {
|
||||
from: "dev-updater-f5".to_string(),
|
||||
// The body, not the wrapper the model is given.
|
||||
text: "Pull before you touch AGENTS.md.".to_string(),
|
||||
turn_start: None,
|
||||
},
|
||||
"{events:?}"
|
||||
);
|
||||
|
||||
@@ -2200,14 +2200,29 @@ async fn pump(
|
||||
// the order the transcript has them -- and because the answer has to
|
||||
// survive being asked a moment later than the driver would have said it.
|
||||
let mut unread: usize = 0;
|
||||
// Where the turn currently running began: the seq of the `Status`
|
||||
// that opened it, which is recorded before any of the turn's own
|
||||
// output. Held here because the pump is the only place that knows a
|
||||
// seq at all, and the only one that sees every driver's turns.
|
||||
let mut turn_start: Option<u64> = None;
|
||||
while let Some(event) = source.recv().await {
|
||||
let ts = now();
|
||||
// Taking a message is how it enters the conversation, and the
|
||||
// conversation is what a phone renders -- so the event becomes the
|
||||
// message here rather than being carried alongside it. One rule
|
||||
// for where a user's message sits: where the session read it.
|
||||
//
|
||||
// A peer message is stamped with the same knowledge for the
|
||||
// opposite reason: it arrives *after* everything it caused, and
|
||||
// the position is the only way a reader can put it back where it
|
||||
// happened -- see `Event::PeerMessage::turn_start`.
|
||||
let event = match event {
|
||||
Event::MessageTaken { id, text, images } => Event::UserMessage { id, text, images },
|
||||
Event::PeerMessage { from, text, .. } => Event::PeerMessage {
|
||||
from,
|
||||
text,
|
||||
turn_start,
|
||||
},
|
||||
other => other,
|
||||
};
|
||||
// Where the session row's figure comes from. Kept here rather than
|
||||
@@ -2264,6 +2279,20 @@ async fn pump(
|
||||
}
|
||||
*shared.last_activity.lock().unwrap() = ts;
|
||||
*shared.written.lock().unwrap() += 1;
|
||||
// The turn's own first line, kept for whatever arrives at
|
||||
// the end of it needing to say where it started. Only the
|
||||
// *opening* status counts: a turn that pauses for a
|
||||
// question or a compaction and resumes is still the turn
|
||||
// that began where it began.
|
||||
match &entry.event {
|
||||
Event::Status {
|
||||
state: SessionStatus::Running,
|
||||
} if turn_start.is_none() => turn_start = Some(entry.seq),
|
||||
Event::Status {
|
||||
state: SessionStatus::Idle | SessionStatus::Exited,
|
||||
} => turn_start = None,
|
||||
_ => {}
|
||||
}
|
||||
// The boundary a held command was waiting for, and the one
|
||||
// place that sees every driver's. Done after the status is
|
||||
// recorded, so the command that runs next sees an idle
|
||||
@@ -2423,6 +2452,62 @@ mod tests {
|
||||
assert!(matches!(events.try_recv(), Ok(Event::CommandSent { .. })));
|
||||
}
|
||||
|
||||
/// A peer message is stamped with where its turn began, so a phone can
|
||||
/// draw it above the reply it caused rather than below it.
|
||||
///
|
||||
/// The live CLI reveals the message only on the turn's `result`, and an
|
||||
/// append-only transcript cannot go back and insert it -- so the
|
||||
/// position has to travel with the event. Without it the note is drawn
|
||||
/// at the end of the turn, which reads as an answer printed above its
|
||||
/// question.
|
||||
#[tokio::test]
|
||||
async fn a_peer_message_carries_the_seq_its_turn_started_at() {
|
||||
let dir = tempfile::tempdir().expect("tempdir");
|
||||
let config_path = dir.path().join("config.ron");
|
||||
let data_dir = dir.path().join("sessions");
|
||||
seed_echo_only(&config_path);
|
||||
let manager = SessionManager::new(config_path, data_dir.clone(), data_dir.join("models"))
|
||||
.expect("manager");
|
||||
let info = manager.spawn_session(echo_spec()).expect("spawn");
|
||||
let session = manager.session(&info.id).expect("live");
|
||||
let mut rx = session.subscribe();
|
||||
|
||||
session.send_message("/peer-turn".to_string(), Vec::new());
|
||||
let seen = collect_until(&mut rx, |event| matches!(event, Event::PeerMessage { .. })).await;
|
||||
|
||||
let opened = seen
|
||||
.iter()
|
||||
.find(|entry| {
|
||||
matches!(
|
||||
entry.event,
|
||||
Event::Status {
|
||||
state: SessionStatus::Running
|
||||
}
|
||||
)
|
||||
})
|
||||
.expect("the turn's opening status")
|
||||
.seq;
|
||||
let note = seen.last().expect("the peer message");
|
||||
let Event::PeerMessage { turn_start, .. } = ¬e.event else {
|
||||
unreachable!()
|
||||
};
|
||||
assert_eq!(*turn_start, Some(opened), "{seen:?}");
|
||||
// And it is worth stamping only because it is genuinely behind the
|
||||
// turn it explains.
|
||||
assert!(note.seq > opened, "{seen:?}");
|
||||
|
||||
// A message that opened no turn is left where it arrived: an
|
||||
// import replays those in place already.
|
||||
session.send_message("/peer".to_string(), Vec::new());
|
||||
let alone =
|
||||
collect_until(&mut rx, |event| matches!(event, Event::PeerMessage { .. })).await;
|
||||
let Some(Event::PeerMessage { turn_start, .. }) = alone.last().map(|entry| &entry.event)
|
||||
else {
|
||||
unreachable!()
|
||||
};
|
||||
assert_eq!(*turn_start, None, "{alone:?}");
|
||||
}
|
||||
|
||||
/// A command waits for the *driver* to be between turns, not for the
|
||||
/// recorded status to say idle.
|
||||
///
|
||||
|
||||
Reference in new issue
Block a user