Hare/yoi
1
1
Fork 0
Code Actions Packages Releases Activity

feat: discover subject session history

Browse Source
This commit is contained in:
Hare
2026-10-02 16:41:29 +09:00
parent 43e3b42055
commit ee6aa9de41
34 changed files with 5167 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/worker-runtime/src/http_server.rs
+3
View File
@@ -2312,6 +2312,9 @@ fn required_runtime_permission(method: &Method, path: &str) -> Option<&'static s
if path.starts_with("/v1/workers/") && *method == Method::DELETE {
return Some("workers:delete");
}
if path.starts_with("/v1/session-public/") && *method == Method::POST {
return Some("workers:read");
}
if path.starts_with("/v1/workers/") && *method == Method::GET {
return Some("workers:read");
}
crates/worker-runtime/src/http_server/runtime_management_api.rs
+74
View File
@@ -289,6 +289,80 @@ impl runtime_api::RuntimeApi for RuntimeManagementApi {
.map_err(api_error)
}
async fn session_public_search(
&self,
request: runtime_api::SessionPublicSearchRequest,
) -> Result<runtime_api::SessionPublicSearchAvailability, runtime_api::RuntimeApiError> {
let scope = auth_workspace_scope(&self.state, auth_extension().as_ref())
.map_err(api_error)?
.ok_or_else(|| {
runtime_api::RuntimeApiError::new(
StatusCode::FORBIDDEN.as_u16(),
"runtime_session_public_scope_required",
"Session public search requires a Workspace-scoped capability",
)
})?;
if scope.workspace_id != request.workspace_id {
return Err(runtime_api::RuntimeApiError::new(
StatusCode::FORBIDDEN.as_u16(),
"runtime_session_public_workspace_scope_mismatch",
"Session public search Workspace scope does not match the authenticated capability",
));
}
#[cfg(feature = "fs-store")]
{
self.state
.runtime
.session_public_search_scoped(&scope, &request)
.map_err(RuntimeHttpRestError::runtime)
.map_err(api_error)
}
#[cfg(not(feature = "fs-store"))]
{
Ok(runtime_api::SessionPublicSearchAvailability::Unavailable {
reason: runtime_api::SessionPublicUnavailableReason::StorageUnavailable,
message: "Session public storage is unavailable".to_string(),
})
}
}
async fn session_public_read(
&self,
request: runtime_api::SessionPublicReadRequest,
) -> Result<runtime_api::SessionPublicReadAvailability, runtime_api::RuntimeApiError> {
let scope = auth_workspace_scope(&self.state, auth_extension().as_ref())
.map_err(api_error)?
.ok_or_else(|| {
runtime_api::RuntimeApiError::new(
StatusCode::FORBIDDEN.as_u16(),
"runtime_session_public_scope_required",
"Session public read requires a Workspace-scoped capability",
)
})?;
if scope.workspace_id != request.workspace_id {
return Err(runtime_api::RuntimeApiError::new(
StatusCode::FORBIDDEN.as_u16(),
"runtime_session_public_workspace_scope_mismatch",
"Session public read Workspace scope does not match the authenticated capability",
));
}
#[cfg(feature = "fs-store")]
{
self.state
.runtime
.session_public_read_scoped(&scope, &request)
.map_err(RuntimeHttpRestError::runtime)
.map_err(api_error)
}
#[cfg(not(feature = "fs-store"))]
{
Ok(runtime_api::SessionPublicReadAvailability::Unavailable {
reason: runtime_api::SessionPublicUnavailableReason::StorageUnavailable,
message: "Session public storage is unavailable".to_string(),
})
}
}
async fn create_worker(
&self,
value: runtime_api::CreateWorkerRequest,
crates/worker-runtime/src/lib.rs
+2
View File
@@ -25,6 +25,8 @@ pub mod resource;
#[cfg(feature = "fs-store")]
pub mod retention;
mod runtime;
#[cfg(feature = "fs-store")]
mod session_public;
pub mod ssh_host_key_probe;
pub mod worker_backend;
pub mod worker_source;
crates/worker-runtime/src/retention.rs
+20
View File
@@ -356,6 +356,26 @@ impl FsWorkerRetentionProvider {
.join(archive_id))
}
pub(crate) fn session_archive(
&self,
archive_id: &str,
) -> Result<(WorkerSessionArchiveManifest, PathBuf), RuntimeError> {
let archive_dir = self.archive_dir(archive_id)?;
let manifest: WorkerSessionArchiveManifest = read_json(
&archive_dir.join("manifest.json"),
"read Worker Session archive",
)?;
if manifest.schema_version != ARCHIVE_SCHEMA_VERSION || manifest.archive_id != archive_id {
return Err(RuntimeError::StoreCorrupt {
operation: "read Worker Session archive",
path: archive_dir.join("manifest.json"),
message: "archive manifest identity mismatch".to_string(),
});
}
let manifest = validate_existing_archive(&archive_dir, &manifest)?;
Ok((manifest, archive_dir.join("session")))
}
fn diagnostics_dir(&self, operation_id: &str) -> Result<PathBuf, RuntimeError> {
validate_id("operation_id", operation_id)?;
Ok(self
crates/worker-runtime/src/runtime.rs
+152
View File
@@ -77,6 +77,41 @@ impl RuntimeWorkspaceScope {
}
const SUBSCRIPTION_QUEUE_CAPACITY: usize = 256;
#[cfg(feature = "fs-store")]
fn session_public_source_unavailable(
source: &runtime_api::SessionPublicSource,
error: &RuntimeError,
) -> Option<(runtime_api::SessionPublicUnavailableReason, &'static str)> {
use runtime_api::SessionPublicUnavailableReason as Reason;
match (source, error) {
(
runtime_api::SessionPublicSource::Retained { .. },
RuntimeError::WorkerNotFound { .. },
) => Some((
Reason::RetentionMissing,
"retained Session storage is missing",
)),
(runtime_api::SessionPublicSource::Archived { .. }, RuntimeError::StoreMissing { .. }) => {
Some((
Reason::ArchiveIncomplete,
"committed Session archive is incomplete",
))
}
(runtime_api::SessionPublicSource::Archived { .. }, RuntimeError::StoreCorrupt { .. }) => {
Some((
Reason::ArchiveIncomplete,
"committed Session archive failed integrity validation",
))
}
(_, RuntimeError::StoreIo { .. } | RuntimeError::StoreCommitOutcomeUnknown { .. }) => {
Some((Reason::StorageUnavailable, "Session storage is unavailable"))
}
_ => None,
}
}
const WORKER_DELETE_FAILURE_DIAGNOSTIC_CODE: &str = "worker_delete_persistence_failed";
const WORKER_DELETE_FAILURE_DIAGNOSTIC_MESSAGE: &str =
"Worker metadata deletion failed; the persisted Worker identity was retained for retry";
@@ -3543,6 +3578,123 @@ impl Runtime {
)
}
/// Search the committed public projection of one Backend-selected retained
/// Session or verified archive without restoring a Worker.
#[cfg(feature = "fs-store")]
pub fn session_public_search_scoped(
&self,
scope: &RuntimeWorkspaceScope,
request: &runtime_api::SessionPublicSearchRequest,
) -> Result<runtime_api::SessionPublicSearchAvailability, RuntimeError> {
let (session_root, archive) =
match self.resolve_session_public_source(scope, &request.source) {
Ok(source) => source,
Err(error) => {
if let Some((reason, message)) =
session_public_source_unavailable(&request.source, &error)
{
return Ok(runtime_api::SessionPublicSearchAvailability::Unavailable {
reason,
message: message.to_string(),
});
}
return Err(error);
}
};
Ok(crate::session_public::search(
&session_root,
request,
archive.as_ref(),
))
}
/// Read one exact public entry from a Backend-selected retained Session or
/// verified archive. No fallback Worker, Segment, or entry is selected.
#[cfg(feature = "fs-store")]
pub fn session_public_read_scoped(
&self,
scope: &RuntimeWorkspaceScope,
request: &runtime_api::SessionPublicReadRequest,
) -> Result<runtime_api::SessionPublicReadAvailability, RuntimeError> {
let (session_root, archive) =
match self.resolve_session_public_source(scope, &request.source) {
Ok(source) => source,
Err(error) => {
if let Some((reason, message)) =
session_public_source_unavailable(&request.source, &error)
{
return Ok(runtime_api::SessionPublicReadAvailability::Unavailable {
reason,
message: message.to_string(),
});
}
return Err(error);
}
};
Ok(crate::session_public::read(
&session_root,
request,
archive.as_ref(),
))
}
#[cfg(feature = "fs-store")]
fn resolve_session_public_source(
&self,
scope: &RuntimeWorkspaceScope,
source: &runtime_api::SessionPublicSource,
) -> Result<
(
std::path::PathBuf,
Option<crate::retention::WorkerSessionArchiveManifest>,
),
RuntimeError,
> {
let state = self.lock()?;
let store = state.fs_store().ok_or_else(|| {
RuntimeError::InvalidRequest(
"Session public reads require an fs-backed Runtime".to_string(),
)
})?;
match source {
runtime_api::SessionPublicSource::Retained { worker_id } => {
let worker_ref = WorkerRef::new(*worker_id);
let worker = state.worker(&worker_ref)?;
if worker.workspace_id.as_deref() != Some(scope.workspace_id.as_str()) {
return Err(RuntimeError::WorkerNotFound {
worker_id: *worker_id,
});
}
Ok((
store
.runtime_dir()
.join("workers")
.join(worker_id.to_string())
.join("session"),
None,
))
}
runtime_api::SessionPublicSource::Archived { archive_id } => {
let runtime_id = state.runtime_identity.as_deref().ok_or_else(|| {
RuntimeError::InvalidRequest(
"Runtime identity is not bound for Session archive reads".to_string(),
)
})?;
let provider = FsWorkerRetentionProvider::new(store.runtime_dir());
let (manifest, session_root) = provider.session_archive(archive_id)?;
if manifest.workspace_id != scope.workspace_id
|| manifest.source_runtime_id != runtime_id
{
return Err(RuntimeError::InvalidRequest(
"Session archive scope does not match the authenticated Runtime capability"
.to_string(),
));
}
Ok((session_root, Some(manifest)))
}
}
}
/// Enumerate host-authoritative Runtime inventory for Backend orphan
/// reconciliation. Runtime identity and Workspace scope are derived here,
/// not accepted in a diagnostic payload.
crates/worker-runtime/src/session_public.rs
+357
View File
@@ -0,0 +1,357 @@
use std::path::Path;
use runtime_api::{
SessionPublicEntryKind, SessionPublicLineage, SessionPublicLineageKind,
SessionPublicReadAvailability, SessionPublicReadMode, SessionPublicReadPage,
SessionPublicReadRequest, SessionPublicSearchAvailability, SessionPublicSearchItem,
SessionPublicSearchPage, SessionPublicSearchRequest, SessionPublicToolPart,
SessionPublicUnavailableReason, WorkerSessionArchiveManifest,
};
use session_store::{
SessionPublicIndex, SessionPublicIndexEntry, SessionPublicIndexEntryKind,
SessionPublicIndexLimits, SessionPublicIndexLineage, SessionPublicIndexOriginKind,
SessionPublicIndexReadError, SessionPublicIndexToolPart, read_session_public_index,
};
use crate::retention::WorkerSessionArchiveManifest as DomainArchiveManifest;
pub(crate) fn search(
session_root: &Path,
request: &SessionPublicSearchRequest,
archive_manifest: Option<&DomainArchiveManifest>,
) -> SessionPublicSearchAvailability {
if request.limit == 0 || request.limit > 100 {
return unavailable(
SessionPublicUnavailableReason::ResourceLimit,
"Session public search limit must be between 1 and 100",
);
}
let index = match load_index(
session_root,
request.max_scan_bytes,
request.max_segments,
request.max_entries,
) {
Ok(index) => index,
Err((reason, message)) => return unavailable(reason, message),
};
if index.session_id != request.expected_session_id {
return unavailable(
SessionPublicUnavailableReason::NotFound,
"Session identity does not match the authorized source",
);
}
if request
.expected_generation
.as_deref()
.is_some_and(|expected| expected != index.generation)
{
return unavailable(
SessionPublicUnavailableReason::InvalidCursor,
"Session public search generation changed",
);
}
let query = request
.query
.as_deref()
.map(str::trim)
.filter(|value| !value.is_empty())
.map(str::to_lowercase);
let mut matches = Vec::new();
for segment in &index.segments {
for entry in &segment.entries {
if !matches_entry(entry, request, query.as_deref()) {
continue;
}
matches.push(SessionPublicSearchItem {
segment_id: segment.segment_id.clone(),
entry_ref: entry.entry_ref.clone(),
kind: entry_kind(entry.kind),
origin: entry.provenance.clone(),
tool_name: entry.tool_name.clone(),
tool_part: entry.tool_part.map(tool_part),
compact: entry.compact_text.clone(),
lineage: lineage(&segment.lineage),
});
}
}
if request.offset > matches.len() {
return unavailable(
SessionPublicUnavailableReason::InvalidCursor,
"Session public search position is stale",
);
}
let end = request
.offset
.saturating_add(request.limit)
.min(matches.len());
let items = matches[request.offset..end].to_vec();
let has_more = end < matches.len();
SessionPublicSearchAvailability::Page {
page: SessionPublicSearchPage {
session_id: index.session_id,
generation: index.generation,
items,
next_offset: has_more.then_some(end),
has_more,
scanned_bytes: index.scanned_bytes,
scanned_segments: index.segments.len(),
scanned_entries: index.scanned_entries,
archive_manifest: archive_manifest.map(api_archive_manifest),
},
}
}
pub(crate) fn read(
session_root: &Path,
request: &SessionPublicReadRequest,
archive_manifest: Option<&DomainArchiveManifest>,
) -> SessionPublicReadAvailability {
if request.max_content_bytes == 0 {
return read_unavailable(
SessionPublicUnavailableReason::ResourceLimit,
"Session public read content limit must be positive",
);
}
let index = match load_index(
session_root,
request.max_scan_bytes,
request.max_segments,
request.max_entries,
) {
Ok(index) => index,
Err((reason, message)) => return read_unavailable(reason, message),
};
if index.session_id != request.expected_session_id {
return read_unavailable(
SessionPublicUnavailableReason::NotFound,
"Session identity does not match the authorized source",
);
}
if request
.expected_generation
.as_deref()
.is_some_and(|expected| expected != index.generation)
{
return read_unavailable(
SessionPublicUnavailableReason::InvalidCursor,
"Session public read generation changed",
);
}
let Some(segment) = index
.segments
.iter()
.find(|segment| segment.segment_id == request.segment_id)
else {
return read_unavailable(
SessionPublicUnavailableReason::NotFound,
"Session entry was not found in the authorized source",
);
};
let mut entries = segment
.entries
.iter()
.filter(|entry| entry.entry_ref == request.entry_ref);
let Some(entry) = entries.next() else {
return read_unavailable(
SessionPublicUnavailableReason::NotFound,
"Session entry was not found in the authorized source",
);
};
if entries.next().is_some() {
return read_unavailable(
SessionPublicUnavailableReason::CorruptLog,
"Session entry identity is ambiguous in its Segment",
);
}
let content = match request.mode {
SessionPublicReadMode::Compact => &entry.compact_text,
SessionPublicReadMode::Full => &entry.full_text,
};
if request.byte_offset > content.len() || !content.is_char_boundary(request.byte_offset) {
return read_unavailable(
SessionPublicUnavailableReason::InvalidCursor,
"Session public read position is stale",
);
}
let mut end = request
.byte_offset
.saturating_add(request.max_content_bytes)
.min(content.len());
while end > request.byte_offset && !content.is_char_boundary(end) {
end -= 1;
}
if end == request.byte_offset && request.byte_offset < content.len() {
return read_unavailable(
SessionPublicUnavailableReason::ResourceLimit,
"Session public read cannot fit one UTF-8 scalar in the content limit",
);
}
let has_more = end < content.len();
SessionPublicReadAvailability::Page {
page: SessionPublicReadPage {
session_id: index.session_id,
generation: index.generation,
segment_id: segment.segment_id.clone(),
entry_ref: entry.entry_ref.clone(),
kind: entry_kind(entry.kind),
origin: entry.provenance.clone(),
lineage: lineage(&segment.lineage),
mode: request.mode,
content: content[request.byte_offset..end].to_string(),
next_byte_offset: has_more.then_some(end),
has_more,
scanned_bytes: index.scanned_bytes,
scanned_segments: index.segments.len(),
scanned_entries: index.scanned_entries,
archive_manifest: archive_manifest.map(api_archive_manifest),
},
}
}
fn load_index(
session_root: &Path,
max_bytes: u64,
max_segments: usize,
max_entries: usize,
) -> Result<SessionPublicIndex, (SessionPublicUnavailableReason, &'static str)> {
read_session_public_index(
session_root,
SessionPublicIndexLimits {
max_bytes,
max_segments,
max_entries,
},
)
.map_err(map_index_error)
}
fn matches_entry(
entry: &SessionPublicIndexEntry,
request: &SessionPublicSearchRequest,
query: Option<&str>,
) -> bool {
if request
.kind
.is_some_and(|kind| kind != entry_kind(entry.kind))
{
return false;
}
if request.tool_name.is_some() || request.tool_part != SessionPublicToolPart::Both {
if entry.kind != SessionPublicIndexEntryKind::Tool {
return false;
}
}
if request
.tool_name
.as_deref()
.is_some_and(|name| entry.tool_name.as_deref() != Some(name))
{
return false;
}
if request.tool_part != SessionPublicToolPart::Both
&& entry.tool_part.map(tool_part) != Some(request.tool_part)
{
return false;
}
query.is_none_or(|query| entry.full_text.to_lowercase().contains(query))
}
fn entry_kind(kind: SessionPublicIndexEntryKind) -> SessionPublicEntryKind {
match kind {
SessionPublicIndexEntryKind::User => SessionPublicEntryKind::User,
SessionPublicIndexEntryKind::Assistant => SessionPublicEntryKind::Assistant,
SessionPublicIndexEntryKind::Tool => SessionPublicEntryKind::Tool,
}
}
fn tool_part(part: SessionPublicIndexToolPart) -> SessionPublicToolPart {
match part {
SessionPublicIndexToolPart::Input => SessionPublicToolPart::Input,
SessionPublicIndexToolPart::Output => SessionPublicToolPart::Output,
}
}
fn lineage(value: &SessionPublicIndexLineage) -> SessionPublicLineage {
SessionPublicLineage {
kind: match value.origin_kind {
SessionPublicIndexOriginKind::Root => SessionPublicLineageKind::Root,
SessionPublicIndexOriginKind::Fork => SessionPublicLineageKind::Fork,
SessionPublicIndexOriginKind::Compact => SessionPublicLineageKind::Compact,
},
parent_segment_id: value.parent_segment_id.clone(),
at_turn_index: value.parent_turn_index,
}
}
fn map_index_error(
error: SessionPublicIndexReadError,
) -> (SessionPublicUnavailableReason, &'static str) {
match error {
SessionPublicIndexReadError::MigrationRequired => (
SessionPublicUnavailableReason::MigrationRequired,
"Session storage requires migration",
),
SessionPublicIndexReadError::Missing => (
SessionPublicUnavailableReason::RetentionMissing,
"Session storage is missing",
),
SessionPublicIndexReadError::Corrupt => (
SessionPublicUnavailableReason::CorruptLog,
"Session log is corrupt",
),
SessionPublicIndexReadError::Storage => (
SessionPublicUnavailableReason::StorageUnavailable,
"Session storage is unavailable",
),
SessionPublicIndexReadError::ResourceLimit => (
SessionPublicUnavailableReason::ResourceLimit,
"Session scan exceeds a resource limit",
),
}
}
fn unavailable(
reason: SessionPublicUnavailableReason,
message: impl Into<String>,
) -> SessionPublicSearchAvailability {
SessionPublicSearchAvailability::Unavailable {
reason,
message: message.into(),
}
}
fn read_unavailable(
reason: SessionPublicUnavailableReason,
message: impl Into<String>,
) -> SessionPublicReadAvailability {
SessionPublicReadAvailability::Unavailable {
reason,
message: message.into(),
}
}
fn api_archive_manifest(manifest: &DomainArchiveManifest) -> WorkerSessionArchiveManifest {
WorkerSessionArchiveManifest {
schema_version: manifest.schema_version,
archive_id: manifest.archive_id.clone(),
workspace_id: manifest.workspace_id.clone(),
source_runtime_id: manifest.source_runtime_id.clone(),
source_worker_id: manifest.source_worker_id,
source_session_id: manifest.source_session_id.clone(),
segment_ids: manifest.segment_ids.clone(),
source_created_at: manifest.source_created_at.clone(),
removed_at: manifest.removed_at.clone(),
archived_at_unix_seconds: manifest.archived_at_unix_seconds,
effective_profile: manifest.effective_profile.clone(),
retention_class: manifest.retention_class.clone(),
content_checksum_sha256: manifest.content_checksum_sha256.clone(),
content_bytes: manifest.content_bytes,
content_file_count: manifest.content_file_count,
policy_id: manifest.policy_id.clone(),
policy_revision: manifest.policy_revision,
operation_id: manifest.operation_id.clone(),
input_fingerprint: manifest.input_fingerprint.clone(),
}
}
crates/worker-runtime/src/worker_source.rs
+7
View File
@@ -581,6 +581,13 @@ impl WorkspaceClient for RuntimeOwnedWorkspaceClient {
client.subjektiv_memory_backend(workspace_id, request).await,
)
}
WorkspaceServerOperation::SubjektivSession(request) => {
generated_workspace_response(
client
.subjektiv_session_backend(workspace_id, request)
.await,
)
}
}
})
})