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

fix: bound memory read continuations

Browse Source
This commit is contained in:
Hare
2026-10-02 09:26:53 +09:00
parent f50a0caea9
commit 860c34b8da
5 changed files with 211 additions and 47 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/server-api/src/lib.rs
+10
View File
@@ -4592,6 +4592,11 @@ pub struct SubjektivMemoryReadRequest {
#[serde(default, skip_serializing_if = "Option::is_none")]
#[schemars(range(min = 0, max = 9_007_199_254_740_991_usize))]
pub offset: Option<usize>,
/// UTF-8 byte offset within `offset`'s line. Used only for continuation of
/// one line that exceeds the response byte budget.
#[serde(default, skip_serializing_if = "Option::is_none")]
#[schemars(range(min = 0, max = 9_007_199_254_740_991_usize))]
pub byte_offset: Option<usize>,
#[serde(default, skip_serializing_if = "Option::is_none")]
#[schemars(range(min = 1, max = 1000))]
pub limit: Option<usize>,
@@ -4719,9 +4724,14 @@ pub struct SubjektivMemoryReadResponse {
pub updated_at: String,
#[schemars(range(min = 0, max = 9_007_199_254_740_991_usize))]
pub body_offset: usize,
#[schemars(range(min = 0, max = 9_007_199_254_740_991_usize))]
pub body_byte_offset: usize,
#[serde(default, skip_serializing_if = "Option::is_none")]
#[schemars(range(min = 0, max = 9_007_199_254_740_991_usize))]
pub body_next_offset: Option<usize>,
#[serde(default, skip_serializing_if = "Option::is_none")]
#[schemars(range(min = 0, max = 9_007_199_254_740_991_usize))]
pub body_next_byte_offset: Option<usize>,
pub body_truncated: bool,
/// Candidate ids on this evidence page. The corresponding bounded anchors
/// are returned in `source_candidates` with the same order.
crates/worker/src/feature/builtin/subjektiv_memory.rs
+2 -2
View File
@@ -43,10 +43,10 @@ const PROPOSE_REVISION_TOOL: &str = "SubjektivMemoryProposeRevision";
const COMMIT_HOOK: &str = "stage-explicit-memory-after-commit";
const RETRY_HOOK: &str = "retry-committed-explicit-memory";
const REWRITE_HOOK: &str = "flush-explicit-memory-before-rewrite";
const MAX_EVIDENCE_REFS: usize = 20;
const MAX_EVIDENCE_REFS: usize = 10;
const QUERY_DESCRIPTION: &str = "Search the connected subject's current Memory revisions. Omitted states means active only; results are bounded and cursors are bound to the subject, filters, order, and store snapshot.";
const READ_DESCRIPTION: &str = "Read one current or immutable historical Memory revision for the connected subject, with line-bounded Markdown and paged provenance anchors. Continue partial reads using the returned exact revision.";
const READ_DESCRIPTION: &str = "Read one current or immutable historical Memory revision for the connected subject, with line- and byte-bounded Markdown plus paged provenance anchors. Continue partial reads using the returned exact revision, line offset, and byte offset.";
const LIST_REVISIONS_DESCRIPTION: &str = "List immutable revisions of one connected-subject Memory in descending revision order using snapshot-bound pagination.";
const REMEMBER_DESCRIPTION: &str = "Explicitly stage a new Memory candidate for the connected subject. This never writes confirmed Memory. With no entry_refs, the committed tool call itself becomes model-origin evidence after the run commits and status is pending_commit.";
const PROPOSE_DESCRIPTION: &str = "Stage a typed correction, resolution, retraction, or reopen proposal for an exact current Memory revision. This validates optimistic concurrency now but never creates a confirmed revision.";
crates/workspace-server/src/server.rs
+182 -38
View File
@@ -16775,10 +16775,11 @@ const SUBJEKTIV_QUERY_DEFAULT_LIMIT: usize = 20;
const SUBJEKTIV_QUERY_MAX_LIMIT: usize = 100;
const SUBJEKTIV_BODY_DEFAULT_LINES: usize = 200;
const SUBJEKTIV_BODY_MAX_LINES: usize = 1_000;
const SUBJEKTIV_EVIDENCE_PAGE_SIZE: usize = 20;
const SUBJEKTIV_CANDIDATE_ANCHOR_LIMIT: usize = 20;
const SUBJEKTIV_BODY_MAX_BYTES: usize = 16 * 1024;
const SUBJEKTIV_EVIDENCE_PAGE_SIZE: usize = 1;
const SUBJEKTIV_CANDIDATE_ANCHOR_LIMIT: usize = crate::subjektiv::MAX_STAGING_ANCHORS;
const SUBJEKTIV_ANCHOR_TEXT_MAX_BYTES: usize = 64;
const SUBJEKTIV_NESTED_EVIDENCE_MAX_BYTES: usize = 32 * 1024;
const SUBJEKTIV_NESTED_EVIDENCE_MAX_BYTES: usize = 40 * 1024;
#[derive(Debug, Serialize, Deserialize)]
#[serde(deny_unknown_fields)]
@@ -17003,7 +17004,8 @@ fn subjektiv_memory_read(
.into());
}
let body_offset = input.offset.unwrap_or(0);
if body_offset > 0 && input.revision.is_none() {
let body_byte_offset = input.byte_offset.unwrap_or(0);
if (body_offset > 0 || body_byte_offset > 0) && input.revision.is_none() {
return Err(Error::InvalidInput(
"body continuation requires the exact revision returned by the first read".into(),
)
@@ -17013,9 +17015,28 @@ fn subjektiv_memory_read(
if body_offset > lines.len() {
return Err(Error::InvalidInput("body line offset exceeds document length".into()).into());
}
let body_end = body_offset.saturating_add(body_limit).min(lines.len());
let body_truncated = body_end < lines.len();
let body_md = lines[body_offset..body_end].concat();
if body_offset == lines.len() && body_byte_offset > 0 {
return Err(Error::InvalidInput("body byte offset exceeds document length".into()).into());
}
if let Some(line) = lines.get(body_offset)
&& (body_byte_offset > line.len() || !line.is_char_boundary(body_byte_offset))
{
return Err(Error::InvalidInput(
"body byte offset must be a UTF-8 boundary within the selected line".into(),
)
.into());
}
let (body_md, body_next) = bounded_memory_body_page(
&lines,
body_offset,
body_byte_offset,
body_limit,
SUBJEKTIV_BODY_MAX_BYTES,
);
let body_truncated = body_next.is_some();
let (body_next_offset, body_next_byte_offset) = body_next
.map(|(line, byte)| (Some(line), Some(byte)))
.unwrap_or((None, None));
let evidence_offset = if let Some(cursor) = input.evidence_cursor {
let cursor: SubjektivEvidenceCursor = decode_subjektiv_cursor("evidence", &cursor)?;
@@ -17124,7 +17145,9 @@ fn subjektiv_memory_read(
created_at: record.created_at,
updated_at: record.updated_at,
body_offset,
body_next_offset: body_truncated.then_some(body_end),
body_byte_offset,
body_next_offset,
body_next_byte_offset,
body_truncated,
source_candidate_ids,
source_candidates,
@@ -17509,6 +17532,45 @@ fn api_proposal_intent(
}
}
fn bounded_memory_body_page(
lines: &[&str],
line_offset: usize,
byte_offset: usize,
line_limit: usize,
byte_limit: usize,
) -> (String, Option<(usize, usize)>) {
let line_end = line_offset.saturating_add(line_limit).min(lines.len());
let mut line_index = line_offset;
let mut within_line = byte_offset;
let mut body = String::new();
while line_index < line_end {
let line = lines[line_index];
let remainder = &line[within_line..];
let available = byte_limit.saturating_sub(body.len());
if remainder.len() <= available {
body.push_str(remainder);
line_index += 1;
within_line = 0;
continue;
}
let mut take = available.min(remainder.len());
while take > 0 && !remainder.is_char_boundary(take) {
take -= 1;
}
if take == 0 {
return (body, Some((line_index, within_line)));
}
body.push_str(&remainder[..take]);
within_line += take;
return (body, Some((line_index, within_line)));
}
if line_index < lines.len() {
(body, Some((line_index, 0)))
} else {
(body, None)
}
}
fn memory_body_lines(body: &str) -> Vec<&str> {
if body.is_empty() {
Vec::new()
@@ -17520,31 +17582,15 @@ fn memory_body_lines(body: &str) -> Vec<&str> {
fn bound_subjektiv_nested_evidence(
candidates: &mut [server_api::SubjektivMemoryEvidenceCandidate],
) -> ApiResult<()> {
loop {
let serialized = serde_json::to_vec(candidates)
.map_err(|error| Error::Store(format!("encode subjektiv evidence page: {error}")))?;
if serialized.len() <= SUBJEKTIV_NESTED_EVIDENCE_MAX_BYTES {
return Ok(());
}
let mut removed = false;
for candidate in candidates.iter_mut().rev() {
if candidate.source_refs.pop().is_some() {
candidate.source_refs_truncated = true;
removed = true;
break;
}
if candidate.evidence.pop().is_some() {
candidate.evidence_truncated = true;
removed = true;
break;
}
}
if !removed {
return Err(Error::Store(
"subjektiv evidence metadata exceeds its response byte budget".into(),
)
.into());
}
let serialized = serde_json::to_vec(candidates)
.map_err(|error| Error::Store(format!("encode subjektiv evidence page: {error}")))?;
if serialized.len() <= SUBJEKTIV_NESTED_EVIDENCE_MAX_BYTES {
Ok(())
} else {
Err(Error::Store(
"subjektiv evidence page exceeds its enforced response byte budget".into(),
)
.into())
}
}
@@ -17603,6 +17649,16 @@ fn bounded_subjektiv_evidence_origin(
}
fn bounded_utf8_bytes(value: String, max_bytes: usize) -> String {
let value = value
.chars()
.map(|character| {
if character.is_control() || matches!(character, '"' | '\\') {
' '
} else {
character
}
})
.collect::<String>();
if value.len() <= max_bytes {
return value;
}
@@ -35402,6 +35458,7 @@ mod tests {
memory_id: first.id.clone(),
revision: None,
offset: None,
byte_offset: None,
limit: Some(1),
evidence_cursor: None,
},
@@ -35430,6 +35487,7 @@ mod tests {
memory_id: first.id.clone(),
revision: None,
offset: initial_read.body_next_offset,
byte_offset: initial_read.body_next_byte_offset,
limit: Some(1),
evidence_cursor: None,
},
@@ -35444,6 +35502,7 @@ mod tests {
memory_id: first.id.clone(),
revision: Some(1),
offset: Some(1),
byte_offset: None,
limit: Some(1),
evidence_cursor: None,
},
@@ -35455,7 +35514,55 @@ mod tests {
assert_eq!(read.revision, 1);
assert_eq!(read.current_revision, 2);
let nested_count = SUBJEKTIV_CANDIDATE_ANCHOR_LIMIT + 5;
let oversized_body = "界".repeat(7_000);
let oversized_memory = store
.create_memory(
&subject.id,
crate::subjektiv::MemoryDraft::active(
memory::extract::CandidateKind::Lesson,
"Page oversized lines by byte",
oversized_body.clone(),
"Keeps tool JSON intact",
"oversized line regression",
),
)
.unwrap();
let oversized_first = subjektiv_memory_read(
&store,
&subject.id,
server_api::SubjektivMemoryReadRequest {
memory_id: oversized_memory.id.clone(),
revision: None,
offset: None,
byte_offset: None,
limit: Some(1),
evidence_cursor: None,
},
)
.unwrap();
assert!(oversized_first.body_md.len() <= SUBJEKTIV_BODY_MAX_BYTES);
assert!(oversized_first.body_truncated);
assert_eq!(oversized_first.body_next_offset, Some(0));
let oversized_second = subjektiv_memory_read(
&store,
&subject.id,
server_api::SubjektivMemoryReadRequest {
memory_id: oversized_memory.id,
revision: Some(oversized_first.revision),
offset: oversized_first.body_next_offset,
byte_offset: oversized_first.body_next_byte_offset,
limit: Some(1),
evidence_cursor: None,
},
)
.unwrap();
assert!(!oversized_second.body_truncated);
assert_eq!(
format!("{}{}", oversized_first.body_md, oversized_second.body_md),
oversized_body
);
let nested_count = SUBJEKTIV_CANDIDATE_ANCHOR_LIMIT;
let nested_evidence = (0..nested_count)
.map(|index| memory::extract::StagingEvidence {
id: format!("nested-{index}-{}", "i".repeat(220)),
@@ -35506,6 +35613,43 @@ mod tests {
..Default::default()
})
.collect::<Vec<_>>();
let mut rejected_evidence = nested_evidence.clone();
let mut extra_evidence = rejected_evidence[0].clone();
extra_evidence.id = "nested-over-limit".to_string();
rejected_evidence.push(extra_evidence);
let mut rejected_refs = nested_refs.clone();
let mut extra_ref = rejected_refs[0].clone();
extra_ref.evidence_id = Some("nested-over-limit".to_string());
rejected_refs.push(extra_ref);
let rejected_candidate = memory::extract::StagingRecord::from_candidate(
"candidate-over-anchor-limit",
"over-anchor-limit-dedup",
memory::schema::SourceRef {
segment_id: "segment-nested".to_string(),
range: [0, nested_count as u64],
},
memory::extract::ExtractedCandidate {
kind: memory::extract::CandidateKind::Lesson,
claim: "Reject unpageable provenance".to_string(),
why_useful: "Every staging path shares the read bound".to_string(),
staleness: None,
evidence_ids: rejected_evidence
.iter()
.map(|item| item.id.clone())
.collect(),
},
rejected_evidence,
rejected_refs,
);
let rejected = store.stage_candidate(crate::subjektiv::SubjectStagingRecord::attach(
&subject.id,
rejected_candidate,
));
assert!(matches!(
rejected,
Err(crate::subjektiv::SubjektivError::InvalidRecord(message))
if message.contains("limited to")
));
let nested_candidate = memory::extract::StagingRecord::from_candidate(
"candidate-nested",
"nested-dedup",
@@ -35551,6 +35695,7 @@ mod tests {
memory_id: nested_memory.id,
revision: Some(1),
offset: None,
byte_offset: None,
limit: None,
evidence_cursor: None,
},
@@ -35560,11 +35705,10 @@ mod tests {
let nested = &nested_read.source_candidates[0];
assert_eq!(nested.evidence.len(), SUBJEKTIV_CANDIDATE_ANCHOR_LIMIT);
assert_eq!(nested.evidence_total, nested_count);
assert!(nested.evidence_truncated);
assert!(!nested.source_refs.is_empty());
assert!(nested.source_refs.len() <= SUBJEKTIV_CANDIDATE_ANCHOR_LIMIT);
assert!(!nested.evidence_truncated);
assert_eq!(nested.source_refs.len(), SUBJEKTIV_CANDIDATE_ANCHOR_LIMIT);
assert_eq!(nested.source_refs_total, nested_count);
assert!(nested.source_refs_truncated);
assert!(!nested.source_refs_truncated);
assert!(nested.evidence.iter().all(|item| {
item.excerpt
.as_ref()
crates/workspace-server/src/subjektiv.rs
+7
View File
@@ -22,6 +22,7 @@ use crate::feature_storage::{
pub const SUBJEKTIV_SCHEMA_VERSION: u32 = 1;
pub const SUBJEKTIV_FEATURE_ID: &str = "subjektiv";
pub const MAX_STAGING_ANCHORS: usize = 10;
pub type Result<T> = std::result::Result<T, SubjektivError>;
@@ -1898,6 +1899,12 @@ fn validate_staging_record(workspace_id: &str, record: &SubjectStagingRecord) ->
));
}
if record.evidence.len() > MAX_STAGING_ANCHORS || record.source_refs.len() > MAX_STAGING_ANCHORS
{
return Err(SubjektivError::InvalidRecord(format!(
"staging evidence/source_refs are limited to {MAX_STAGING_ANCHORS} items each"
)));
}
let mut evidence_ids = HashSet::new();
for evidence in &record.evidence {
validate_label("evidence id", &evidence.id)?;
docs/design/subjektiv-store.md
+10 -7
View File
@@ -184,15 +184,18 @@ page members.
Read accepts a positive exact revision or resolves the current revision once. A
missing historical revision never falls back to current. Markdown pagination is
line-based (default 200, maximum 1000) and reports `body_truncated` plus
`body_next_offset`; every nonzero continuation offset must also supply the exact
line-based (default 200, maximum 1000) with an additional 16 KiB UTF-8 response
budget. It reports `body_truncated`, `body_next_offset`, and
`body_next_byte_offset`; an oversized single line resumes from the returned byte
boundary within that same line. Every continuation offset must also supply the exact
revision returned by the first page, so a current-revision change cannot mix body
versions. Provenance uses a separate immutable-revision-bound cursor and returns at
most 20 candidate/derivation references per page. Expanded candidates share one
page-wide budget of 20 evidence records and 20 source references, plus a hard 32 KiB
versions. Provenance uses a separate immutable-revision-bound cursor and returns one
candidate/derivation reference per page. A staged candidate contains at most 10
evidence records and 10 source references, all returned together under a hard 40 KiB
serialized budget for the expanded candidate array; displayed anchor text is capped
at 64 UTF-8 bytes and total/truncation metadata reports omissions.
Explicit staging rejects larger anchor sets. Candidate evidence is the bounded host-resolved anchor saved in staging; raw Session bodies are not copied
at 64 UTF-8 bytes. Every staging path rejects larger anchor sets, so accepted
provenance is never made unreachable. Candidate evidence is the bounded
host-resolved anchor saved in staging; raw Session bodies are not copied
into Memory responses. Resolved, retracted, and historical revisions remain
addressable by ID.