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
3 changes: 3 additions & 0 deletions src/apps/cli/src/modes/chat/session_lineage.rs
Original file line number Diff line number Diff line change
Expand Up @@ -1239,6 +1239,7 @@ mod session_lineage_tests {
user_input: "continue".to_string(),
original_user_input: None,
user_message_metadata: None,
session_binding: None,
},
);
assert_eq!(
Expand Down Expand Up @@ -1290,6 +1291,7 @@ mod session_lineage_tests {
user_input: "continue".to_string(),
original_user_input: None,
user_message_metadata: None,
session_binding: None,
},
AgenticEvent::TextChunk {
session_id: "first".to_string(),
Expand Down Expand Up @@ -1525,6 +1527,7 @@ mod session_lineage_tests {
user_input: "question".to_string(),
original_user_input: None,
user_message_metadata: None,
session_binding: None,
},
false,
);
Expand Down
27 changes: 23 additions & 4 deletions src/crates/assembly/core/src/agentic/coordination/coordinator.rs
Original file line number Diff line number Diff line change
Expand Up @@ -4689,6 +4689,7 @@ Update the persona files and delete BOOTSTRAP.md as soon as bootstrap is complet
user_input: prompt.clone(),
original_user_input: (display_input != prompt).then_some(display_input),
user_message_metadata: Some(user_message_metadata.clone()),
session_binding: self.session_manager.session_workspace_binding(&session_id),
})
.await;
self.emit_event(AgenticEvent::ModelRoundStarted {
Expand Down Expand Up @@ -6213,6 +6214,7 @@ Update the persona files and delete BOOTSTRAP.md as soon as bootstrap is complet
user_input: MANUAL_COMPACTION_COMMAND.to_string(),
original_user_input: None,
user_message_metadata: user_message_metadata.clone(),
session_binding: self.session_manager.session_workspace_binding(&session_id),
})
.await;

Expand Down Expand Up @@ -7469,6 +7471,7 @@ Update the persona files and delete BOOTSTRAP.md as soon as bootstrap is complet
None
},
user_message_metadata: user_message_metadata.clone(),
session_binding: self.session_manager.session_workspace_binding(&session_id),
})
.await;

Expand Down Expand Up @@ -11225,6 +11228,7 @@ Update the persona files and delete BOOTSTRAP.md as soon as bootstrap is complet
user_input: user_input_text.clone(),
original_user_input: None,
user_message_metadata: None,
session_binding: self.session_manager.session_workspace_binding(&session_id),
})
.await;
}
Expand Down Expand Up @@ -15984,6 +15988,9 @@ impl openbitfun_runtime_ports::AgentUserShellCommandPort for ConversationCoordin
user_input: format!("!{}", request.command),
original_user_input: None,
user_message_metadata: None,
session_binding: self
.session_manager
.session_workspace_binding(&request.session_id),
};

let session_manager = Arc::clone(&self.session_manager);
Expand Down Expand Up @@ -21791,10 +21798,22 @@ pub(crate) mod tests {
.expect("load turns");
let turn = turns.last().expect("shell turn");
let events = coordinator.event_queue.dequeue_batch(100).await;
assert!(events.iter().any(|envelope| matches!(
&envelope.event,
AgenticEvent::DialogTurnStarted { turn_id, .. } if turn_id == "turn-shell"
)));
let started_binding = events.iter().find_map(|envelope| match &envelope.event {
AgenticEvent::DialogTurnStarted {
turn_id,
session_binding,
..
} if turn_id == "turn-shell" => Some(session_binding.clone()),
_ => None,
});
// A surface that has not loaded the session attributes the Turn to
// its owning workspace from this binding, not to its current one.
assert_eq!(
started_binding
.expect("turn started event")
.and_then(|binding| binding.workspace_path),
session.config.workspace_path
);
assert!(!events.iter().any(|envelope| matches!(
&envelope.event,
AgenticEvent::DialogTurnFailed { turn_id, .. } if turn_id == "turn-shell"
Expand Down
19 changes: 19 additions & 0 deletions src/crates/assembly/core/src/agentic/session/session_manager.rs
Original file line number Diff line number Diff line change
Expand Up @@ -3017,6 +3017,25 @@ impl SessionManager {
self.sessions.get(session_id).map(|s| s.clone())
}

/// Owning workspace facts for Turn lifecycle events, read without cloning
/// the session. `None` when this host has not loaded the session.
pub fn session_workspace_binding(
&self,
session_id: &str,
) -> Option<openbitfun_events::SessionWorkspaceBinding> {
self.sessions.get(session_id).map(|session| {
let config = &session.config;
openbitfun_events::SessionWorkspaceBinding {
workspace_path: config.workspace_path.clone(),
project_workspace_path: config.project_workspace_path.clone(),
execution_target: config.execution_target.clone(),
workspace_id: config.workspace_id.clone(),
remote_connection_id: config.remote_connection_id.clone(),
remote_ssh_host: config.remote_ssh_host.clone(),
}
})
}

/// Read selector facts without restoring history, reserving runtime capacity,
/// or waiting for an executing Turn's mutation permit. An omitted locator
/// supports legacy callers only when this host already owns the session.
Expand Down
24 changes: 24 additions & 0 deletions src/crates/contracts/events/src/agentic.rs
Original file line number Diff line number Diff line change
Expand Up @@ -94,6 +94,27 @@ pub struct DeepReviewQueueState {
pub session_concurrency_high: bool,
}

/// Workspace binding of the session that owns a Turn. A surface that has not
/// loaded the session (remote control, a bot, another device) uses it to
/// attribute the Turn to its owning workspace instead of guessing one.
#[derive(Debug, Clone, Default, Serialize, Deserialize, PartialEq, Eq)]
pub struct SessionWorkspaceBinding {
/// Execution root of the session.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub workspace_path: Option<String>,
/// Main project root that owns persistence for the session.
#[serde(default, skip_serializing_if = "Option::is_none")]
pub project_workspace_path: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub execution_target: Option<SessionExecutionTarget>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub workspace_id: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub remote_connection_id: Option<String>,
#[serde(default, skip_serializing_if = "Option::is_none")]
pub remote_ssh_host: Option<String>,
}

#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(tag = "type")]
pub enum AgenticEvent {
Expand Down Expand Up @@ -182,6 +203,9 @@ pub enum AgenticEvent {
original_user_input: Option<String>,
/// Image metadata JSON for UI rendering (id, name, data_url, mime_type, image_path)
user_message_metadata: Option<serde_json::Value>,
/// Owning workspace of the session. Absent from older hosts.
#[serde(default, skip_serializing_if = "Option::is_none")]
session_binding: Option<SessionWorkspaceBinding>,
},

/// Low-frequency linking event that associates a hidden subagent session
Expand Down
84 changes: 79 additions & 5 deletions src/crates/contracts/events/src/frontend_projection.rs
Original file line number Diff line number Diff line change
Expand Up @@ -85,17 +85,41 @@ pub fn project_agentic_frontend_event(event: AgenticEvent) -> Option<AgenticFron
user_input,
original_user_input,
user_message_metadata,
} => Some(AgenticFrontendEvent::new(
"agentic://dialog-turn-started",
json!({
session_binding,
} => {
let mut payload = json!({
"sessionId": session_id,
"turnId": turn_id,
"turnIndex": turn_index,
"userInput": user_input,
"originalUserInput": original_user_input,
"userMessageMetadata": user_message_metadata,
}),
)),
});
// Same keys as `agentic://session-created`, so a surface that has
// not loaded the session attributes it to its owning workspace.
if let (Some(binding), Some(fields)) = (session_binding, payload.as_object_mut()) {
let binding_fields = [
("workspacePath", json!(binding.workspace_path)),
(
"projectWorkspacePath",
json!(binding.project_workspace_path),
),
("executionTarget", json!(binding.execution_target)),
("workspaceId", json!(binding.workspace_id)),
("remoteConnectionId", json!(binding.remote_connection_id)),
("remoteSshHost", json!(binding.remote_ssh_host)),
];
for (key, value) in binding_fields {
if !value.is_null() {
fields.insert(key.to_string(), value);
}
}
}
Some(AgenticFrontendEvent::new(
"agentic://dialog-turn-started",
payload,
))
}
AgenticEvent::SubagentSessionLinked {
session_id,
subagent_dialog_turn_id,
Expand Down Expand Up @@ -607,6 +631,56 @@ mod tests {
assert_eq!(projected.payload["workspaceId"], "workspace-wt-1");
}

#[test]
fn dialog_turn_started_projects_owning_workspace_binding() {
let projected = project_agentic_frontend_event(AgenticEvent::DialogTurnStarted {
session_id: "session-1".to_string(),
turn_id: "turn-1".to_string(),
turn_index: 20,
user_input: "continue".to_string(),
original_user_input: None,
user_message_metadata: None,
session_binding: Some(crate::SessionWorkspaceBinding {
workspace_path: Some("D:/repo".to_string()),
project_workspace_path: Some("D:/repo".to_string()),
execution_target: None,
workspace_id: Some("workspace-repo".to_string()),
remote_connection_id: None,
remote_ssh_host: None,
}),
})
.expect("projected");

assert_eq!(projected.event_name, "agentic://dialog-turn-started");
assert_eq!(projected.payload["workspacePath"], "D:/repo");
assert_eq!(projected.payload["projectWorkspacePath"], "D:/repo");
assert_eq!(projected.payload["workspaceId"], "workspace-repo");
let fields = projected.payload.as_object().expect("object payload");
assert!(!fields.contains_key("executionTarget"));
assert!(!fields.contains_key("remoteConnectionId"));
}

#[test]
fn dialog_turn_started_without_binding_keeps_legacy_payload() {
// An older host serializes the event without `session_binding`.
let legacy = serde_json::json!({
"type": "DialogTurnStarted",
"session_id": "session-1",
"turn_id": "turn-1",
"turn_index": 0,
"user_input": "hello",
"original_user_input": null,
"user_message_metadata": null,
});
let event: AgenticEvent = serde_json::from_value(legacy).expect("legacy event");
let projected = project_agentic_frontend_event(event).expect("projected");

let fields = projected.payload.as_object().expect("object payload");
assert!(!fields.contains_key("workspacePath"));
assert!(!fields.contains_key("workspaceId"));
assert_eq!(projected.payload["sessionId"], "session-1");
}

#[test]
fn thinking_chunk_projects_to_text_chunk_event() {
let projected = project_agentic_frontend_event(AgenticEvent::ThinkingChunk {
Expand Down
3 changes: 2 additions & 1 deletion src/crates/contracts/events/src/lib.rs
Original file line number Diff line number Diff line change
Expand Up @@ -15,7 +15,8 @@ pub mod types;
pub use agentic::{
AgenticEvent, AgenticEventEnvelope, AgenticEventPriority, DeepReviewQueueReason,
DeepReviewQueueState, DeepReviewQueueStatus, ModelRoundAttemptDiagnostic,
ModelRoundAttemptToolDiagnostic, SubagentParentInfo, ToolEventData, ToolEventIdentity,
ModelRoundAttemptToolDiagnostic, SessionWorkspaceBinding, SubagentParentInfo, ToolEventData,
ToolEventIdentity,
};
pub use backend::{
BackgroundCommandLifecycleInfo, ToolExecutionCompletedInfo, ToolExecutionErrorInfo,
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -736,6 +736,7 @@ mod tests {
user_input: "prompt".to_string(),
original_user_input: None,
user_message_metadata: None,
session_binding: None,
}
}

Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -6290,6 +6290,7 @@ mod tests {
user_input: "hidden".to_string(),
original_user_input: None,
user_message_metadata: None,
session_binding: None,
});
tracker.mark_persistence_clean();
let previous_version = tracker.version();
Expand Down Expand Up @@ -6380,6 +6381,7 @@ mod tests {
user_input: "hello".to_string(),
original_user_input: None,
user_message_metadata: None,
session_binding: None,
});
tracker.handle_agentic_event(&AgenticEvent::DialogTurnCompleted {
session_id: "session-a".to_string(),
Expand Down Expand Up @@ -6448,6 +6450,7 @@ mod tests {
user_input: "hello".to_string(),
original_user_input: None,
user_message_metadata: None,
session_binding: None,
});
tracker.handle_agentic_event(&AgenticEvent::DialogTurnCompleted {
session_id: "session-a".to_string(),
Expand Down Expand Up @@ -6538,6 +6541,7 @@ mod tests {
user_input: "hello".to_string(),
original_user_input: None,
user_message_metadata: None,
session_binding: None,
});
tracker.handle_agentic_event(&AgenticEvent::DialogTurnFailed {
session_id: "session-a".to_string(),
Expand Down Expand Up @@ -6567,6 +6571,7 @@ mod tests {
user_input: "queued follow-up".to_string(),
original_user_input: None,
user_message_metadata: None,
session_binding: None,
});
tracker.handle_agentic_event(&AgenticEvent::TextChunk {
session_id: "session-a".to_string(),
Expand Down
Loading
Loading