1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
use serde::Deserialize;
// Stream event types now live in agent-core so rpc_dispatch can use them
// without an upward dependency on the engine layer.
pub use agent_core::{AgentEvent, LlmEvent, SessionEvent, StreamEvent};
/// Per-turn signals shared by the host and the engine.
///
/// Engine → host: whether the turn reached its NORMAL end — the final
/// response came back un-cancelled and asked for no tools (nothing left to
/// do). Set before the engine publishes that turn's final `MessageHistory`.
/// `Done` cannot say this: several cancelled paths also end `Ok` + `Done`.
/// A host that cancels a turn reads it once the stream is drained: a turn
/// that finished just before the cancel reached it (Esc a moment too late)
/// is complete, not interrupted.
///
/// Host → engine: WHY the host cancelled the turn, noted before it cancels
/// the token, so a canceled `tool_result` says so ("Canceled by user" only
/// when it was the user). Nothing noted = the turn's default cause: the
/// user (the historical wording), or what the host chose at turn start
/// (`with_default_cause`: a driver turn whose token something else cancels,
/// e.g. the grant's deadline, was cut by the driver ending).
#[derive(Debug, Clone, Default)]
pub struct TurnCompletion {
completed: std::sync::Arc<std::sync::atomic::AtomicBool>,
/// `0` = unset, else `CancelCause as u8`.
cancel_cause: std::sync::Arc<std::sync::atomic::AtomicU8>,
/// Used when nothing was noted; `None` = the user.
default_cause: Option<CancelCause>,
}
/// Why a host cancelled a turn, as far as a canceled tool result says.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[repr(u8)]
pub enum CancelCause {
User = 1,
CostCap = 2,
Restart = 3,
Host = 4,
Driver = 5,
}
impl CancelCause {
/// Leading words of a canceled `tool_result`. Every variant starts with
/// "Canceled" (the phase-4 bound tests key on it); the user's is the
/// historical, byte-identical "Canceled by user".
pub fn phrase(self) -> &'static str {
match self {
CancelCause::User => "Canceled by user",
CancelCause::CostCap => "Canceled (session cost cap reached)",
CancelCause::Restart => "Canceled (Synaps restarted)",
CancelCause::Host => "Canceled (session stopped by the host)",
CancelCause::Driver => "Canceled (session driver revoked)",
}
}
fn from_u8(v: u8) -> Option<Self> {
Some(match v {
1 => CancelCause::User,
2 => CancelCause::CostCap,
3 => CancelCause::Restart,
4 => CancelCause::Host,
5 => CancelCause::Driver,
_ => return None,
})
}
}
impl TurnCompletion {
pub fn new() -> Self {
Self::default()
}
/// A turn whose un-noted cancels are `cause`'s (see the type docs).
pub fn with_default_cause(cause: CancelCause) -> Self {
Self {
default_cause: Some(cause),
..Self::default()
}
}
pub fn completed(&self) -> bool {
self.completed.load(std::sync::atomic::Ordering::Acquire)
}
pub(crate) fn mark_completed(&self) {
self.completed
.store(true, std::sync::atomic::Ordering::Release);
}
/// Note why the host is cancelling this turn. First cause wins: a
/// teardown that revokes a driver and then cancels keeps its own cause.
pub fn note_cancel_cause(&self, cause: CancelCause) {
let _ = self.cancel_cause.compare_exchange(
0,
cause as u8,
std::sync::atomic::Ordering::AcqRel,
std::sync::atomic::Ordering::Acquire,
);
}
/// Leading words of this turn's canceled tool results.
pub(crate) fn cancel_phrase(&self) -> &'static str {
CancelCause::from_u8(self.cancel_cause.load(std::sync::atomic::Ordering::Acquire))
.or(self.default_cause)
.unwrap_or(CancelCause::User)
.phrase()
}
}
/// Shared mutable auth state. Lives behind `Arc<RwLock<_>>` so the spawned
/// streaming task and the parent Runtime always see the same (freshest) token.
#[derive(Clone)]
pub(super) struct AuthState {
pub(super) auth_token: String,
pub(super) auth_type: String,
pub(super) refresh_token: Option<String>,
pub(super) token_expires: Option<u64>,
/// Local source only: mtime of auth.json observed just before the token
/// in `auth_token` was loaded. A different on-disk mtime means the file
/// was rotated externally (`synaps login`, cron, sync) and the fast path
/// must reload. `None` disables the check (remote / never loaded).
pub(super) auth_file_mtime: Option<std::time::SystemTime>,
/// Which credential `auth_token` was vended for:
/// `"<local | remote:<endpoint>#<principal fingerprint>>|<storage_key>"`
/// (see `runtime/auth.rs::anthropic_binding`). The principal fingerprint
/// is an opaque SHA-256 prefix of the machine token — never the token
/// itself, never an access token. `None` = unknown/legacy (api_key
/// harnesses, scrubbed). The pre-stream refresh refuses to serve a
/// cached token whose binding no longer matches the selected account on
/// the current source, so a config/source/principal switch can never
/// silently keep the previous account's token.
pub(super) bound_credential: Option<String>,
}
/// Redacting Debug — never print the access or refresh token.
impl std::fmt::Debug for AuthState {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("AuthState")
.field(
"auth_token",
&if self.auth_token.is_empty() {
"<empty>"
} else {
"<redacted>"
},
)
.field("auth_type", &self.auth_type)
.field(
"refresh_token",
&self.refresh_token.as_ref().map(|_| "<redacted>"),
)
.field("token_expires", &self.token_expires)
.field("auth_file_mtime", &self.auth_file_mtime)
.field("bound_credential", &self.bound_credential)
.finish()
}
}
#[allow(dead_code)]
#[derive(Debug, Deserialize)]
pub(super) struct AnthropicAuth {
#[serde(rename = "type")]
pub(super) auth_type: String,
pub(super) refresh: Option<String>,
pub(super) access: Option<String>,
pub(super) expires: Option<u64>,
pub(super) key: Option<String>,
}