feat(session-limits): LS7 — câblage backend app-tauri (taps niveaux 1&2 + reprise annulable)
Branche le SessionLimitService dans l'application Tauri et expose la surface de reprise/annulation au front. - application/agent/lifecycle.rs : LaunchAgentOutput.profile exposé (None sur réattache/idempotent, Some sur lancement effectif). - application/terminal/registry.rs : StructuredSessions::meta_for_session() (lookup agent/node par SessionId pour le tap niveau 1). - app-tauri/state.rs : ResumeContext(s), AppAgentResumer (impl du port AgentResumer au-dessus de LaunchAgent), instanciation + câblage du SessionLimitService (TokioScheduler + drain des réveils) dans AppState::build. - app-tauri/commands.rs : taps niveau 1 (agent_send) et niveau 2 (launch_agent, parser regex confiné), alimentation de resume_contexts, commande cancel_resume. - app-tauri/lib.rs : enregistrement de cancel_resume dans le handler. - app-tauri/Cargo.toml : dépendance async-trait. Tests : session_limit_wiring.rs (2 tests de composition) + meta_for_session dans structured_registry_d1.rs ; fixtures dto_agents/dto_chat ajustées (profile: None). Tout vert. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
@ -1137,6 +1137,11 @@ pub async fn launch_agent(
|
||||
// artefacts from previous AppImages do not survive into this activation.
|
||||
state.reconcile_claude_run_dirs(&project).await;
|
||||
|
||||
// Session-limit resume context (LS7, §21.5): record the Project + cell size keyed by
|
||||
// agent so an auto-resume (which only carries agent/node/conversation) can recompose a
|
||||
// full `LaunchAgentInput`. Cloned here — `project` is moved into the launch below.
|
||||
let resume_project = project.clone();
|
||||
|
||||
let output = state
|
||||
.launch_agent
|
||||
.execute(LaunchAgentInput {
|
||||
@ -1153,7 +1158,21 @@ pub async fn launch_agent(
|
||||
.await
|
||||
.map_err(ErrorDto::from)?;
|
||||
|
||||
if let Ok(mut contexts) = state.resume_contexts.lock() {
|
||||
contexts.insert(
|
||||
agent_id,
|
||||
crate::state::ResumeContext {
|
||||
project: resume_project,
|
||||
rows: request.rows,
|
||||
cols: request.cols,
|
||||
},
|
||||
);
|
||||
}
|
||||
|
||||
let session_id = output.session.id;
|
||||
// Host cell of the freshly (re)launched session — the pivot the level-2 tap reports
|
||||
// to the service alongside the agent.
|
||||
let host_node_id = output.session.node_id;
|
||||
|
||||
// §17.4/§17.6: a structured launch routes to an `AgentSession`, registered
|
||||
// **only** in `StructuredSessions` — its id is never a live PTY. Such a cell is
|
||||
@ -1166,6 +1185,18 @@ pub async fn launch_agent(
|
||||
// Register the xterm output channel for this session.
|
||||
let gen = state.pty_bridge.register(session_id, on_output);
|
||||
|
||||
// Level-2 session-limit detector (§21, niveau 2 — PTY/TUI sans adapter
|
||||
// structuré). Selection rule (§21.10-4) is the single infra source `applies`:
|
||||
// arm a parser **only** for a profile without a structured adapter that declares
|
||||
// a `rate_limit_pattern` (anti-double-détection avec le niveau 1). A `None`
|
||||
// profile (reattach/idempotent) or an invalid regex ⇒ no parser, jamais de panique.
|
||||
let rate_limit_parser = output
|
||||
.profile
|
||||
.as_ref()
|
||||
.filter(|p| infrastructure::ratelimit::applies(p))
|
||||
.and_then(|p| p.rate_limit_pattern.as_ref())
|
||||
.and_then(infrastructure::RateLimitParser::new);
|
||||
|
||||
// Subscribe to the PTY's byte stream and pump it to the channel.
|
||||
// The stream is a blocking iterator; it runs on a dedicated OS thread and
|
||||
// ends when the PTY hits EOF or this attach is superseded.
|
||||
@ -1173,8 +1204,30 @@ pub async fn launch_agent(
|
||||
match state.pty_port.subscribe_output(&handle) {
|
||||
Ok(stream) => {
|
||||
let bridge: std::sync::Arc<PtyBridge> = std::sync::Arc::clone(&state.pty_bridge);
|
||||
// Captured for the level-2 tap (moved into the pump thread).
|
||||
let service = std::sync::Arc::clone(&state.session_limit_service);
|
||||
let detect_clock = std::sync::Arc::new(infrastructure::SystemClock::new());
|
||||
let conversation_id = request.conversation_id.clone();
|
||||
std::thread::spawn(move || {
|
||||
for chunk in stream {
|
||||
// §21.10-4 tap (best-effort par fragment) : avant de consommer le
|
||||
// fragment, on cherche le motif de limite. Une détection arme la
|
||||
// reprise via le service (détecter→planifier). Dormant si pas de
|
||||
// parser. La fragmentation PTV (motif coupé entre deux fragments)
|
||||
// est un raté best-effort connu pour LS7.
|
||||
if let Some(parser) = &rate_limit_parser {
|
||||
let text = String::from_utf8_lossy(&chunk);
|
||||
if let Some(limit) =
|
||||
parser.detect(&text, domain::ports::Clock::now_millis(&*detect_clock))
|
||||
{
|
||||
service.on_rate_limited(
|
||||
agent_id,
|
||||
host_node_id,
|
||||
conversation_id.clone(),
|
||||
limit.resets_at_ms,
|
||||
);
|
||||
}
|
||||
}
|
||||
if !bridge.send_output(&session_id, chunk) {
|
||||
break;
|
||||
}
|
||||
@ -1279,8 +1332,23 @@ pub async fn agent_send(
|
||||
// `Final` event (or stream end / superseded channel), then detaches *only its
|
||||
// own* generation so a concurrent re-attach is never torn down.
|
||||
let bridge = std::sync::Arc::clone(&state.chat_bridge);
|
||||
// Level-1 session-limit tap (§21, niveau 1 — structuré). Resolve the agent + host
|
||||
// cell of this session once; a `RateLimited` turn event then feeds the service
|
||||
// (détecter→planifier). DORMANT en composition B-2 (aucune session structurée
|
||||
// vivante) mais câblé pour forward-compat. `conversation_id` non disponible ici ⇒
|
||||
// `None` (acceptable LS7).
|
||||
let service = std::sync::Arc::clone(&state.session_limit_service);
|
||||
let meta = state.structured_sessions.meta_for_session(&sid);
|
||||
std::thread::spawn(move || {
|
||||
for event in stream {
|
||||
// Non-terminal, sans contenu chat : un signal de limite alimente le service
|
||||
// (le badge UI vient du bus `AgentRateLimited`, pas du flux chat) puis on
|
||||
// continue à drainer comme pour un battement.
|
||||
if let domain::ports::ReplyEvent::RateLimited { resets_at_ms } = &event {
|
||||
if let Some((agent_id, node_id)) = meta {
|
||||
service.on_rate_limited(agent_id, node_id, None, *resets_at_ms);
|
||||
}
|
||||
}
|
||||
// Heartbeats carry no chat content (readiness/heartbeat lot 1) ⇒ no wire
|
||||
// chunk; skip them while still draining so the turn runs to its `Final`.
|
||||
let Some(chunk) = crate::chat::chunk_from_event(event) else {
|
||||
@ -1298,6 +1366,26 @@ pub async fn agent_send(
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// `cancel_resume` — annule la **reprise automatique** armée pour un agent limité
|
||||
/// (ARCHITECTURE §21.1-4, fenêtre annulable).
|
||||
///
|
||||
/// Délègue à [`SessionLimitService::cancel_resume`](application::SessionLimitService::cancel_resume) :
|
||||
/// désarme le réveil et, **seulement si** l'annulation a réussi (le réveil n'avait pas
|
||||
/// encore tiré), publie `AgentResumeCancelled`. Renvoie `true` ssi une reprise a
|
||||
/// effectivement été annulée (`false` si aucune n'était armée, ou si elle venait de
|
||||
/// tirer — auquel cas la reprise suit son cours).
|
||||
///
|
||||
/// # Errors
|
||||
/// Returns an [`ErrorDto`] (`INVALID` for a malformed agent id).
|
||||
#[tauri::command]
|
||||
pub async fn cancel_resume(
|
||||
agent_id: String,
|
||||
state: State<'_, AppState>,
|
||||
) -> Result<bool, ErrorDto> {
|
||||
let id = parse_agent_id(&agent_id)?;
|
||||
Ok(state.session_limit_service.cancel_resume(id))
|
||||
}
|
||||
|
||||
/// `interrupt_agent` — the **Interrompre** path (cadrage C4 §4.2).
|
||||
///
|
||||
/// Routes to [`OrchestratorService::interrupt_agent`], which `preempt`s the agent's
|
||||
|
||||
Reference in New Issue
Block a user