fix: complete paste artifact storage contract
This commit is contained in:
@@ -193,6 +193,42 @@ impl WorkerEvent {
|
||||
/// variants — emits an alert and inserts a `[unknown input segment]`
|
||||
/// placeholder into the LLM context so neither user nor LLM is blind to
|
||||
/// the dropped intent.
|
||||
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
|
||||
#[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
|
||||
#[cfg_attr(feature = "json-schema", derive(schemars::JsonSchema))]
|
||||
#[serde(rename_all = "snake_case")]
|
||||
pub enum PasteArtifactMediaType {
|
||||
TextPlainUtf8,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
|
||||
#[cfg_attr(feature = "typescript", derive(ts_rs::TS))]
|
||||
#[cfg_attr(feature = "json-schema", derive(schemars::JsonSchema))]
|
||||
#[serde(rename_all = "snake_case")]
|
||||
pub enum PasteArtifactAvailability {
|
||||
Available,
|
||||
Unavailable,
|
||||
IntegrityFailed,
|
||||
}
|
||||
|
||||
impl PasteArtifactMediaType {
|
||||
pub fn as_str(self) -> &'static str {
|
||||
match self {
|
||||
Self::TextPlainUtf8 => "text/plain; charset=utf-8",
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl PasteArtifactAvailability {
|
||||
pub fn as_str(self) -> &'static str {
|
||||
match self {
|
||||
Self::Available => "available",
|
||||
Self::Unavailable => "unavailable",
|
||||
Self::IntegrityFailed => "integrity_failed",
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Session-owned reference to a large pasted-input artifact.
|
||||
///
|
||||
/// The reference contains only bounded integrity and provenance metadata. The
|
||||
@@ -203,6 +239,11 @@ impl WorkerEvent {
|
||||
#[cfg_attr(feature = "json-schema", derive(schemars::JsonSchema))]
|
||||
pub struct PasteArtifactRef {
|
||||
pub artifact_id: String,
|
||||
pub created_at_ms: u64,
|
||||
pub media_type: PasteArtifactMediaType,
|
||||
/// Availability observed when this immutable reference was committed.
|
||||
/// Reads revalidate storage and integrity rather than trusting this field.
|
||||
pub availability: PasteArtifactAvailability,
|
||||
pub byte_len: u64,
|
||||
pub char_count: u64,
|
||||
pub line_count: u64,
|
||||
@@ -275,11 +316,14 @@ impl Segment {
|
||||
use std::fmt::Write as _;
|
||||
let _ = write!(
|
||||
out,
|
||||
"[Large paste stored as artifact {}: {} bytes, {} chars, {} lines, sha256 {}; use SearchInputArtifact and ReadInputArtifact to inspect it]",
|
||||
"[Large paste stored as artifact {}: {} bytes, {} chars, {} lines, {}, {}, created at {} ms, sha256 {}; use SearchInputArtifact and ReadInputArtifact to inspect it]",
|
||||
artifact.artifact_id,
|
||||
artifact.byte_len,
|
||||
artifact.char_count,
|
||||
artifact.line_count,
|
||||
artifact.media_type.as_str(),
|
||||
artifact.availability.as_str(),
|
||||
artifact.created_at_ms,
|
||||
artifact.sha256
|
||||
);
|
||||
}
|
||||
@@ -1239,6 +1283,9 @@ mod tests {
|
||||
fn paste_artifact_segment_roundtrips_without_body() {
|
||||
let artifact = PasteArtifactRef {
|
||||
artifact_id: "019ca7c8-57b6-7f05-8edf-524147aba7b2".to_string(),
|
||||
created_at_ms: 1_700_000_000_000,
|
||||
media_type: PasteArtifactMediaType::TextPlainUtf8,
|
||||
availability: PasteArtifactAvailability::Available,
|
||||
byte_len: 65_536,
|
||||
char_count: 65_530,
|
||||
line_count: 200,
|
||||
|
||||
@@ -7,11 +7,11 @@ use crate::{
|
||||
CommandStreamSlice, CompactionLifecycle, CompactionLifecycleState, CompletionEntry,
|
||||
CompletionKind, ErrorCode, Event, Greeting, InFlightBlock, InFlightSnapshot,
|
||||
InFlightToolCallState, InternalWorkerKind, InternalWorkerRef, InternalWorkerSnapshot,
|
||||
InvokeKind, MemoryWorkerEvent, Method, PasteArtifactRef, Permission, RewindSummary,
|
||||
RewindTarget, RewindTargetId, RunResult, ScopeRule, Segment, SessionContentPart,
|
||||
SessionEntryProvenance, SessionMessageRole, SessionSnapshot, SessionSnapshotEntry,
|
||||
SessionSnapshotEntryData, SessionToolAttachment, ToolResultDisposition, TurnResult,
|
||||
WorkerEvent, WorkerStatus,
|
||||
InvokeKind, MemoryWorkerEvent, Method, PasteArtifactAvailability, PasteArtifactMediaType,
|
||||
PasteArtifactRef, Permission, RewindSummary, RewindTarget, RewindTargetId, RunResult,
|
||||
ScopeRule, Segment, SessionContentPart, SessionEntryProvenance, SessionMessageRole,
|
||||
SessionSnapshot, SessionSnapshotEntry, SessionSnapshotEntryData, SessionToolAttachment,
|
||||
ToolResultDisposition, TurnResult, WorkerEvent, WorkerStatus,
|
||||
subscription::{
|
||||
EventSubscriptionSelector, SubscriptionEvent, SubscriptionEventPayload, SubscriptionFrame,
|
||||
SubscriptionFramePayload, SubscriptionId, SubscriptionRejectionCode, SubscriptionRequest,
|
||||
@@ -79,6 +79,8 @@ pub fn generated_protocol_types() -> String {
|
||||
push_decl::<Greeting>(&cfg, &mut output);
|
||||
push_decl::<Alert>(&cfg, &mut output);
|
||||
push_decl::<MemoryWorkerEvent>(&cfg, &mut output);
|
||||
push_decl::<PasteArtifactMediaType>(&cfg, &mut output);
|
||||
push_decl::<PasteArtifactAvailability>(&cfg, &mut output);
|
||||
push_decl::<PasteArtifactRef>(&cfg, &mut output);
|
||||
push_decl::<Segment>(&cfg, &mut output);
|
||||
push_decl::<WorkerEvent>(&cfg, &mut output);
|
||||
|
||||
@@ -8,6 +8,7 @@ license.workspace = true
|
||||
[dependencies]
|
||||
base64.workspace = true
|
||||
agen = { workspace = true }
|
||||
fs4.workspace = true
|
||||
serde = { workspace = true, features = ["derive"] }
|
||||
serde_json = { workspace = true }
|
||||
sha2.workspace = true
|
||||
|
||||
@@ -450,6 +450,15 @@ mod tests {
|
||||
.unwrap();
|
||||
|
||||
assert_eq!(reference.byte_len, content.len() as u64);
|
||||
assert!(reference.created_at_ms > 0);
|
||||
assert_eq!(
|
||||
reference.media_type,
|
||||
protocol::PasteArtifactMediaType::TextPlainUtf8
|
||||
);
|
||||
assert_eq!(
|
||||
reference.availability,
|
||||
protocol::PasteArtifactAvailability::Available
|
||||
);
|
||||
assert_eq!(reference.char_count, content.chars().count() as u64);
|
||||
assert_eq!(reference.source_entry_id, "entry-1");
|
||||
assert_eq!(
|
||||
@@ -490,6 +499,55 @@ mod tests {
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn concurrent_paste_writes_atomically_enforce_aggregate_caps() {
|
||||
let tmp = tempfile::TempDir::new().unwrap();
|
||||
let session_id = new_session_id();
|
||||
let barrier = std::sync::Arc::new(std::sync::Barrier::new(3));
|
||||
let limits = PasteArtifactLimits {
|
||||
max_artifact_bytes: 4,
|
||||
max_session_bytes: 8,
|
||||
max_session_artifacts: 1,
|
||||
};
|
||||
let mut handles = Vec::new();
|
||||
for entry_id in ["entry-1", "entry-2"] {
|
||||
let root = tmp.path().to_path_buf();
|
||||
let barrier = barrier.clone();
|
||||
handles.push(std::thread::spawn(move || {
|
||||
let store = FsStore::new(root).unwrap();
|
||||
barrier.wait();
|
||||
store.write_paste_artifact(session_id, entry_id, "1234", limits)
|
||||
}));
|
||||
}
|
||||
barrier.wait();
|
||||
let results = handles
|
||||
.into_iter()
|
||||
.map(|handle| handle.join().unwrap())
|
||||
.collect::<Vec<_>>();
|
||||
assert_eq!(results.iter().filter(|result| result.is_ok()).count(), 1);
|
||||
assert_eq!(
|
||||
results
|
||||
.iter()
|
||||
.filter(|result| matches!(result, Err(StoreError::PasteArtifactLimit(_))))
|
||||
.count(),
|
||||
1
|
||||
);
|
||||
assert_eq!(
|
||||
std::fs::read_dir(
|
||||
FsStore::new(tmp.path())
|
||||
.unwrap()
|
||||
.paste_artifact_dir(session_id)
|
||||
)
|
||||
.unwrap()
|
||||
.filter_map(Result::ok)
|
||||
.filter(
|
||||
|entry| entry.path().extension().and_then(|value| value.to_str()) == Some("json")
|
||||
)
|
||||
.count(),
|
||||
1
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn paste_artifact_limits_and_corruption_fail_closed() {
|
||||
let tmp = tempfile::TempDir::new().unwrap();
|
||||
@@ -498,6 +556,7 @@ mod tests {
|
||||
let limits = PasteArtifactLimits {
|
||||
max_artifact_bytes: 5,
|
||||
max_session_bytes: 8,
|
||||
max_session_artifacts: 2,
|
||||
};
|
||||
let first = store
|
||||
.write_paste_artifact(session_id, "entry-1", "1234", limits)
|
||||
|
||||
@@ -3,8 +3,10 @@
|
||||
use std::fs;
|
||||
use std::io::Write as _;
|
||||
use std::path::Path;
|
||||
use std::time::{SystemTime, UNIX_EPOCH};
|
||||
|
||||
use protocol::PasteArtifactRef;
|
||||
use fs4::fs_std::FileExt;
|
||||
use protocol::{PasteArtifactAvailability, PasteArtifactMediaType, PasteArtifactRef};
|
||||
use serde::{Deserialize, Serialize};
|
||||
use sha2::{Digest, Sha256};
|
||||
|
||||
@@ -15,6 +17,7 @@ use crate::StoreError;
|
||||
pub struct PasteArtifactLimits {
|
||||
pub max_artifact_bytes: u64,
|
||||
pub max_session_bytes: u64,
|
||||
pub max_session_artifacts: u64,
|
||||
}
|
||||
|
||||
impl Default for PasteArtifactLimits {
|
||||
@@ -22,6 +25,7 @@ impl Default for PasteArtifactLimits {
|
||||
Self {
|
||||
max_artifact_bytes: 8 * 1024 * 1024,
|
||||
max_session_bytes: 64 * 1024 * 1024,
|
||||
max_session_artifacts: 1_024,
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -48,7 +52,14 @@ pub(crate) fn write_to_dir(
|
||||
)));
|
||||
}
|
||||
fs::create_dir_all(artifact_dir)?;
|
||||
let aggregate_lock = fs::OpenOptions::new()
|
||||
.create(true)
|
||||
.read(true)
|
||||
.write(true)
|
||||
.open(artifact_dir.join(".aggregate.lock"))?;
|
||||
FileExt::lock_exclusive(&aggregate_lock)?;
|
||||
let mut aggregate = 0_u64;
|
||||
let mut artifact_count = 0_u64;
|
||||
for entry in fs::read_dir(artifact_dir)? {
|
||||
let path = entry?.path();
|
||||
if path.extension().and_then(|value| value.to_str()) != Some("json") {
|
||||
@@ -56,6 +67,9 @@ pub(crate) fn write_to_dir(
|
||||
}
|
||||
let stored: StoredPasteArtifact = serde_json::from_slice(&fs::read(&path)?)?;
|
||||
verify(&stored, &stored.reference.artifact_id)?;
|
||||
artifact_count = artifact_count.checked_add(1).ok_or_else(|| {
|
||||
StoreError::PasteArtifactLimit("session artifact count overflow".to_string())
|
||||
})?;
|
||||
aggregate = aggregate
|
||||
.checked_add(stored.reference.byte_len)
|
||||
.ok_or_else(|| {
|
||||
@@ -71,10 +85,23 @@ pub(crate) fn write_to_dir(
|
||||
limits.max_session_bytes
|
||||
)));
|
||||
}
|
||||
if artifact_count >= limits.max_session_artifacts {
|
||||
return Err(StoreError::PasteArtifactLimit(format!(
|
||||
"session already has {artifact_count} artifacts; maximum is {}",
|
||||
limits.max_session_artifacts
|
||||
)));
|
||||
}
|
||||
|
||||
let artifact_id = uuid::Uuid::now_v7().to_string();
|
||||
let created_at_ms = SystemTime::now()
|
||||
.duration_since(UNIX_EPOCH)
|
||||
.map_err(|error| StoreError::PasteArtifactIntegrity(error.to_string()))?
|
||||
.as_millis() as u64;
|
||||
let reference = PasteArtifactRef {
|
||||
artifact_id: artifact_id.clone(),
|
||||
created_at_ms,
|
||||
media_type: PasteArtifactMediaType::TextPlainUtf8,
|
||||
availability: PasteArtifactAvailability::Available,
|
||||
byte_len,
|
||||
char_count: content.chars().count() as u64,
|
||||
line_count: line_count(content),
|
||||
@@ -130,6 +157,9 @@ pub(crate) fn read_from_dir(
|
||||
fn verify(stored: &StoredPasteArtifact, artifact_id: &str) -> Result<(), StoreError> {
|
||||
let actual_digest = sha256_hex(&stored.content);
|
||||
if stored.reference.artifact_id != artifact_id
|
||||
|| stored.reference.created_at_ms == 0
|
||||
|| stored.reference.media_type != PasteArtifactMediaType::TextPlainUtf8
|
||||
|| stored.reference.availability != PasteArtifactAvailability::Available
|
||||
|| stored.reference.byte_len != stored.content.len() as u64
|
||||
|| stored.reference.char_count != stored.content.chars().count() as u64
|
||||
|| stored.reference.line_count != line_count(&stored.content)
|
||||
|
||||
+10
-2
@@ -76,8 +76,13 @@ impl Atom {
|
||||
Atom::PasteArtifact(artifact) => Some((
|
||||
Style::default().fg(Color::Magenta),
|
||||
format!(
|
||||
"[Paste artifact {} | {} chars, {} lines]",
|
||||
artifact.artifact_id, artifact.char_count, artifact.line_count
|
||||
"[Paste artifact {} | {} chars, {} lines, {}, {}, created {} ms]",
|
||||
artifact.artifact_id,
|
||||
artifact.char_count,
|
||||
artifact.line_count,
|
||||
artifact.media_type.as_str(),
|
||||
artifact.availability.as_str(),
|
||||
artifact.created_at_ms
|
||||
),
|
||||
)),
|
||||
Atom::FileRef(r) => Some((Style::default().fg(Color::Cyan), r.label())),
|
||||
@@ -932,6 +937,9 @@ mod submit_segments_tests {
|
||||
fn restored_paste_artifact_remains_a_typed_segment() {
|
||||
let artifact = protocol::PasteArtifactRef {
|
||||
artifact_id: "019ca7c8-57b6-7f05-8edf-524147aba7b2".to_string(),
|
||||
created_at_ms: 1_700_000_000_000,
|
||||
media_type: protocol::PasteArtifactMediaType::TextPlainUtf8,
|
||||
availability: protocol::PasteArtifactAvailability::Available,
|
||||
byte_len: 65_536,
|
||||
char_count: 65_530,
|
||||
line_count: 200,
|
||||
|
||||
+14
-4
@@ -1299,8 +1299,13 @@ fn chip_span_for(seg: &Segment, fallback: Style) -> (Style, String) {
|
||||
Segment::PasteArtifact { artifact } => (
|
||||
Style::default().fg(Color::Magenta),
|
||||
format!(
|
||||
"[Paste artifact {} | {} chars, {} lines]",
|
||||
artifact.artifact_id, artifact.char_count, artifact.line_count
|
||||
"[Paste artifact {} | {} chars, {} lines, {}, {}, created {} ms]",
|
||||
artifact.artifact_id,
|
||||
artifact.char_count,
|
||||
artifact.line_count,
|
||||
artifact.media_type.as_str(),
|
||||
artifact.availability.as_str(),
|
||||
artifact.created_at_ms
|
||||
),
|
||||
),
|
||||
Segment::FileRef { path } => (Style::default().fg(Color::Cyan), format!("@{path}")),
|
||||
@@ -1322,8 +1327,13 @@ fn segment_display_text(seg: &Segment) -> String {
|
||||
id, chars, lines, ..
|
||||
} => format!("[Clipboard #{id} | {chars} chars, {lines} lines]"),
|
||||
Segment::PasteArtifact { artifact } => format!(
|
||||
"[Paste artifact {} | {} chars, {} lines]",
|
||||
artifact.artifact_id, artifact.char_count, artifact.line_count
|
||||
"[Paste artifact {} | {} chars, {} lines, {}, {}, created {} ms]",
|
||||
artifact.artifact_id,
|
||||
artifact.char_count,
|
||||
artifact.line_count,
|
||||
artifact.media_type.as_str(),
|
||||
artifact.availability.as_str(),
|
||||
artifact.created_at_ms
|
||||
),
|
||||
Segment::FileRef { path } => format!("@{path}"),
|
||||
Segment::Flow { selector } => format!("[Flow: {selector}]"),
|
||||
|
||||
Reference in New Issue
Block a user