Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
28 commits
Select commit Hold shift + click to select a range
8867d36
chore(vendor): bump submodule pointers
senamakel Oct 13, 2026
19f1bfe
chore(vendor): bump tiny* submodule pointers
senamakel Oct 13, 2026
9aa7d0e
chore(vendor): bump tinyagents submodule
senamakel Oct 13, 2026
0992d1d
Merge upstream/main
senamakel Oct 13, 2026
ceafa5f
chore(vendor): bump tinyflows submodule
senamakel Oct 13, 2026
9888b77
chore(vendor): bump tinyflows submodule
senamakel Oct 13, 2026
7e27cea
chore: files changed Cargo.lock
senamakel Oct 13, 2026
5a8aa42
fix(threads): purge session store data when deleting threads
senamakel Oct 13, 2026
364044d
chore(threads): remove unused turn_state imports
senamakel Oct 13, 2026
a5b6e49
chore(core): remove unused imports across runtime and web chat
senamakel Oct 13, 2026
7043faf
refactor(web_chat): qualify Arc path in progress bridge
senamakel Oct 13, 2026
c1e1632
feat(session_store): persist agent sessions to workspace
senamakel Oct 13, 2026
474eb20
feat(agent): add multimodal message support
senamakel Oct 13, 2026
63b9dcd
feat(agent): add multimodal attachment support to agent sessions
senamakel Oct 13, 2026
0cce035
refactor(inference): drop shared chat model adapter
senamakel Oct 13, 2026
5729d3b
fix(threads): preserve mirrored turn state when observations arrive o…
senamakel Oct 13, 2026
ade9529
test(profiles): add host profile and session import tests
senamakel Oct 13, 2026
bda7cdb
feat(web_chat): attach multimodal files to agent turns
senamakel Oct 13, 2026
3fe9f20
test(session_store): cover session store module behaviour
senamakel Oct 13, 2026
6c80973
test(threads): add mirror observe turn state tests
senamakel Oct 13, 2026
7465e1d
chore(deps): drop stale tinystoragedrivers 0.7.0 lock entries
senamakel Oct 13, 2026
035135b
fix(multimodal): prefer blob stash over legacy sidecars for attachments
senamakel Oct 13, 2026
5f26bf2
chore(deps): add anyhow to tinyagents-runtime
senamakel Oct 13, 2026
f216418
chore(web_chat): merge tinyagents_session imports
senamakel Oct 13, 2026
3f8d8ec
test(json_rpc): update turn state seeding to session store api
senamakel Oct 13, 2026
96b1a7b
chore(vendor): bump tinyagents submodule
senamakel Oct 13, 2026
50a665e
refactor(agent): pin tinyagents main and migrate host to TurnStates
senamakel Oct 13, 2026
d36344c
Merge remote-tracking branch 'upstream/main' into tinyagents-main-pin
senamakel Oct 13, 2026
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
1 change: 1 addition & 0 deletions Cargo.lock

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

1 change: 1 addition & 0 deletions crates/openhuman-app/Cargo.lock

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

3 changes: 2 additions & 1 deletion crates/openhuman-core/src/agent/attachments/mod.rs
Original file line number Diff line number Diff line change
Expand Up @@ -458,7 +458,8 @@ pub(crate) async fn has_resolvable_image(
) -> anyhow::Result<bool> {
let rehydrated = crate::agent::multimodal::rehydrate_image_placeholders(&[
tinyagents_session::transcript::TranscriptMessage::user(prompt),
]);
])
.await;
let mut paths = parse(prompt)
.1
.into_iter()
Expand Down
2 changes: 1 addition & 1 deletion crates/openhuman-core/src/agent/harness/graph.rs
Original file line number Diff line number Diff line change
Expand Up @@ -125,7 +125,7 @@ pub(crate) async fn run_channel_turn_via_graph(
}
}
let prepared = if crate::agent::multimodal::has_image_placeholders(history) {
crate::agent::multimodal::rehydrate_image_placeholders(history)
crate::agent::multimodal::rehydrate_image_placeholders(history).await
} else {
history.clone()
};
Expand Down
183 changes: 147 additions & 36 deletions crates/openhuman-core/src/agent/multimodal.rs
Original file line number Diff line number Diff line change
Expand Up @@ -23,7 +23,7 @@ use tinyagents_harness::multimodal::{
markers,
payload::sha256_prefix,
resolve::{resolve_file, resolve_image, TextExtractor},
AttachmentStash,
BlobAttachmentStash,
};

use tinyagents_harness::multimodal::{FilePayload, MultimodalError};
Expand Down Expand Up @@ -438,12 +438,16 @@ pub async fn stash_image_attachments(message: &str, image_config: &MultimodalCon
match resolve_image(reference, &images, max_image_bytes, &client).await {
Ok(data_uri) => {
let id = sha256_prefix(data_uri.as_bytes());
match stash().write(&id, &data_uri).await {
Ok(path) => tracing::debug!(
let written = match stash() {
Some(stash) => stash.write(&id, &data_uri).await,
None => Err(anyhow::anyhow!("no attachment stash for this tenant")),
};
match written {
Ok(stashed) => tracing::debug!(
target: "multimodal",
id = %id,
path = %path.display(),
"[multimodal::images][stash] persisted attachment to disk"
len = stashed.len,
"[multimodal::images][stash] persisted attachment to the blob store"
),
Err(err) => tracing::warn!(
target: "multimodal",
Expand Down Expand Up @@ -508,14 +512,17 @@ fn row_has_image_placeholders(row: &TranscriptMessage) -> bool {
})
}

/// Rehydrate `[Image: … #att:<id>]` placeholders back into local
/// `[IMAGE:<path>]` markers pointing at the on-disk attachment, returning a
/// provider-only copy. Resolution re-reads the file at dispatch. Placeholders
/// whose id is absent (file evicted/swept, or written by a different workspace)
/// keep their text. The attachment host migrates managed sidecars into acting
/// Rehydrate `[Image: … #att:<id>]` placeholders back into `[IMAGE:…]` markers,
/// returning a provider-only copy. An attachment stashed in the blob store
/// comes back as an inline `data:` URI; a legacy sidecar file in the
/// attachments directory as its path (re-read at dispatch). Placeholders whose
/// id is in neither (evicted, or written by a different workspace) keep their
/// text. The attachment host migrates managed sidecars into acting
/// workspace originals before provider resolution, including text fallbacks.
pub fn rehydrate_image_placeholders(messages: &[TranscriptMessage]) -> Vec<TranscriptMessage> {
let index = build_attachment_index();
pub async fn rehydrate_image_placeholders(
messages: &[TranscriptMessage],
) -> Vec<TranscriptMessage> {
let index = build_attachment_index(messages).await;
messages
.iter()
.map(|m| {
Expand All @@ -542,45 +549,69 @@ pub fn rehydrate_image_placeholders(messages: &[TranscriptMessage]) -> Vec<Trans
.collect()
}

// ── The on-disk attachment stash ─────────────────────────────────────────

/// Preserve legacy conversation attachments without size-based eviction.
const ATTACHMENTS_MAX_BYTES: u64 = u64::MAX;
/// Legacy stash references remain durable alongside new workspace uploads.
const ATTACHMENTS_TTL: Duration = Duration::MAX;
// ── The attachment stash ─────────────────────────────────────────────────

/// Process-global on-disk attachments directory. Installed once at core startup
/// via [`init_attachments_dir`].
/// Process-global legacy attachments directory (`<workspace>/attachments`),
/// installed once at core startup via [`init_attachments_dir`]. Sidecar files
/// written before the blob stash stay readable from here.
static ATTACHMENTS_DIR: OnceLock<PathBuf> = OnceLock::new();

/// Install the on-disk attachments directory (`<workspace>/attachments`). Call
/// once at core startup. Idempotent — first writer wins.
/// The workspace whose storage backend holds the blob stash; the parent of the
/// directory given to [`init_attachments_dir`].
static ATTACHMENTS_WORKSPACE: OnceLock<PathBuf> = OnceLock::new();

/// Install the legacy attachments directory (`<workspace>/attachments`) and,
/// from its parent, the workspace the blob stash lives in. Call once at core
/// startup. Idempotent — first writer wins.
pub fn init_attachments_dir(dir: PathBuf) {
if let Some(workspace) = dir.parent() {
let _ = ATTACHMENTS_WORKSPACE.set(workspace.to_path_buf());
}
let _ = ATTACHMENTS_DIR.set(dir);
}

/// Resolve the attachments dir, falling back to a **per-user private** dir when
/// unset (CLI / direct invocation / tests that never called
/// [`init_attachments_dir`]). The persistence-pollution fix and rehydration
/// both hold either way.
/// Resolve the legacy attachments dir, falling back to a **per-user private**
/// dir when unset (CLI / direct invocation / tests that never called
/// [`init_attachments_dir`]).
pub(crate) fn attachments_dir() -> PathBuf {
ATTACHMENTS_DIR
.get()
.cloned()
.unwrap_or_else(fallback_attachments_dir)
}

/// The stash over the resolved attachments directory. The mechanism (atomic
/// dedup'd writes, cap eviction, TTL sweep, index, managed-path check) is
/// `tinyagents_harness::multimodal::AttachmentStash`; the directory, cap and TTL
/// are this host's policy.
fn stash() -> AttachmentStash {
AttachmentStash::new(attachments_dir(), ATTACHMENTS_MAX_BYTES, ATTACHMENTS_TTL)
/// The calling tenant's blob-backed stash: the workspace's storage backend once
/// [`init_attachments_dir`] ran, else a process-local in-memory one (a CLI or
/// test with no workspace: placeholders resolve for the life of the process).
/// The mechanism (dedup'd writes, cap eviction, TTL sweep, id index) is
/// `tinyagents_harness::multimodal::BlobAttachmentStash`; the cap and TTL are
/// this host's policy (none: references stay resolvable).
fn stash() -> Option<BlobAttachmentStash> {
match ATTACHMENTS_WORKSPACE.get() {
Some(workspace) => crate::agent::session_store::attachment_stash_for(workspace),
None => Some(memory_stash().clone()),
}
}

fn memory_stash() -> &'static BlobAttachmentStash {
static MEMORY: OnceLock<BlobAttachmentStash> = OnceLock::new();
MEMORY.get_or_init(|| {
use tinystoragedrivers::StorageBackend as _;
let scoped = tinystoragedrivers::MemoryStorage::new()
.for_scope(&tinystoragedrivers::Scope::local())
.expect("the memory backend binds the local scope");
BlobAttachmentStash::new(
std::sync::Arc::clone(scoped.blobs()),
std::sync::Arc::clone(scoped.documents()),
u64::MAX,
Duration::MAX,
)
})
}

/// Per-user fallback attachments dir used only when [`init_attachments_dir`]
/// was never called. Uses the OS user cache dir (e.g. `~/Library/Caches/…`,
/// `~/.cache/…`) so persisted image bytes aren't dropped into a world-readable
/// `~/.cache/…`) so legacy sidecars aren't looked for in a world-readable
/// shared `temp_dir()` on multi-user hosts. Only falls back to `temp_dir()`
/// when no user cache dir can be resolved at all.
fn fallback_attachments_dir() -> PathBuf {
Expand All @@ -589,9 +620,89 @@ fn fallback_attachments_dir() -> PathBuf {
.unwrap_or_else(|| std::env::temp_dir().join("openhuman-attachments"))
}

/// Build an `id -> path` index from a single read of the attachments dir.
fn build_attachment_index() -> HashMap<String, PathBuf> {
stash().build_index()
/// `id -> marker target` for the placeholders in `messages`: the blob stash's
/// bytes as a `data:` URI, else (for an id the stash lacks) a legacy sidecar
/// file in the attachments directory, as its path. The marker target is carried
/// as a `PathBuf` only because that is the harness's index type; it is rendered
/// verbatim inside `[IMAGE:…]`.
async fn build_attachment_index(messages: &[TranscriptMessage]) -> HashMap<String, PathBuf> {
let mut wanted = Vec::new();
for row in messages {
wanted.extend(placeholder_ids(&row.content));
for part in row.parts.iter().flatten() {
if let TranscriptPart::Text { text } = part {
wanted.extend(placeholder_ids(text));
}
}
}
wanted.sort();
wanted.dedup();
let mut index = HashMap::new();
if wanted.is_empty() {
return index;
}
if let Some(stash) = stash() {
for id in &wanted {
match stash.read_data_uri(id).await {
Ok(Some(data_uri)) => {
index.insert(id.clone(), PathBuf::from(data_uri));
}
Ok(None) => {}
Err(err) => tracing::warn!(
target: "multimodal",
id = %id,
reason = %err,
"[multimodal::images][stash] could not read attachment; trying legacy sidecars"
),
}
}
}
if index.len() < wanted.len() {
for (id, path) in legacy_attachment_index() {
if wanted.contains(&id) {
index.entry(id).or_insert(path);
}
}
}
index
}

/// The `#att:<id>` ids of the sidecar placeholders in `text`.
fn placeholder_ids(text: &str) -> Vec<String> {
markers::extract_image_placeholders_in_text(text)
.iter()
.filter_map(|placeholder| {
let after = placeholder.split(markers::IMAGE_STASH_REF).nth(1)?;
let id = after.trim_end_matches(']').trim();
(!id.is_empty()).then(|| id.to_string())
})
.collect()
}

/// `id -> path` of the legacy sidecar files directly under the attachments
/// directory (written before the blob stash). Symlinks and nested entries are
/// ignored.
fn legacy_attachment_index() -> HashMap<String, PathBuf> {
legacy_attachment_index_in(&attachments_dir())
}

fn legacy_attachment_index_in(dir: &std::path::Path) -> HashMap<String, PathBuf> {
let Ok(entries) = std::fs::read_dir(dir) else {
return HashMap::new();
};
entries
.flatten()
.filter(|entry| entry.file_type().is_ok_and(|kind| kind.is_file()))
.filter_map(|entry| {
let path = entry.path();
let stem = path.file_stem()?.to_str()?.to_string();
let valid = !stem.is_empty()
&& stem
.bytes()
.all(|b| b.is_ascii_alphanumeric() || matches!(b, b'-' | b'_'));
valid.then_some((stem, path))
})
.collect()
}

/// Compatibility no-op: durable conversation attachments are not swept.
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -180,8 +180,8 @@ async fn stash_image_attachments_replaces_marker_with_placeholder() {
}

#[tokio::test]
async fn image_placeholder_rehydrates_to_disk_path_for_provider() {
// Ingress: stash the image to disk and leave a placeholder.
async fn image_placeholder_rehydrates_to_inline_data_uri_for_provider() {
// Ingress: stash the image in the blob store and leave a placeholder.
let msg = format!("describe [IMAGE:{TINY_PNG_DATA_URI}]");
let placeholdered = stash_image_attachments(&msg, &MultimodalConfig::default()).await;
let messages = vec![TranscriptMessage::user(placeholdered)];
Expand All @@ -190,7 +190,7 @@ async fn image_placeholder_rehydrates_to_disk_path_for_provider() {
// Dispatch (vision model): rehydrate to an inline [IMAGE:<path>] marker that
// points at the on-disk attachment. The index is rebuilt from disk on every
// call (no in-memory state), so this resolves even after a process restart.
let rehydrated = rehydrate_image_placeholders(&messages);
let rehydrated = rehydrate_image_placeholders(&messages).await;
assert_eq!(rehydrated.len(), 1);
let content = rehydrated[0].content.clone();
assert!(
Expand All @@ -202,14 +202,10 @@ async fn image_placeholder_rehydrates_to_disk_path_for_provider() {
"placeholder consumed: {content}"
);

// The marker points at a real file on disk.
let start = content.find("[IMAGE:").unwrap() + "[IMAGE:".len();
let end = content[start..].find(']').unwrap() + start;
let path = &content[start..end];
assert!(path.ends_with(".png"), "disk path carries ext: {path}");
// The marker carries the stashed bytes as an inline data URI.
assert!(
std::path::Path::new(path).is_file(),
"attachment persisted to disk: {path}"
content.contains("[IMAGE:data:image/png;base64,"),
"stashed attachment rehydrated inline: {content}"
);

// Round-trip: the provider prep step re-reads the file back into a data URI,
Expand All @@ -230,14 +226,14 @@ async fn image_placeholder_rehydrates_to_disk_path_for_provider() {
);
}

#[test]
fn rehydrate_missing_stash_id_keeps_placeholder_text() {
#[tokio::test]
async fn rehydrate_missing_stash_id_keeps_placeholder_text() {
// A placeholder whose id is absent from the stash (e.g. after a restart) is
// left verbatim rather than dropped — the model still sees a text mention.
let messages = vec![TranscriptMessage::user(
"see [Image: image #att:deadbeefdeadbeef]".to_string(),
)];
let out = rehydrate_image_placeholders(&messages);
let out = rehydrate_image_placeholders(&messages).await;
assert!(out[0]
.content
.contains("[Image: image #att:deadbeefdeadbeef]"));
Expand Down Expand Up @@ -296,7 +292,7 @@ fn extract_image_placeholders_pulls_att_tokens_in_order() {
#[tokio::test]
async fn sweep_keeps_fresh_attachments() {
// A freshly-written attachment (age < TTL) survives the startup sweep, and
// the disk index resolves it — the core of restart-survival.
// the index resolves it — the core of restart-survival.
let msg = format!("[IMAGE:{TINY_PNG_DATA_URI}]");
let placeholdered = stash_image_attachments(&msg, &MultimodalConfig::default()).await;
let id = placeholdered
Expand All @@ -308,9 +304,22 @@ async fn sweep_keeps_fresh_attachments() {

sweep_stale_attachments().await;

let index = build_attachment_index();
let index = build_attachment_index(&[TranscriptMessage::user(placeholdered.clone())]).await;
assert!(
index.contains_key(&id),
"fresh attachment {id} retained after sweep"
);
}

#[test]
fn legacy_sidecar_files_stay_indexed_by_id() {
// Conversations written before the blob stash keep `<id>.<ext>` files in the
// attachments directory; their placeholders must still resolve.
let dir = tempfile::tempdir().unwrap();
std::fs::write(dir.path().join("abc123.png"), b"png").unwrap();
std::fs::create_dir(dir.path().join("nested")).unwrap();
std::fs::write(dir.path().join("bad name.png"), b"png").unwrap();
let index = legacy_attachment_index_in(dir.path());
assert_eq!(index.len(), 1, "{index:?}");
assert_eq!(index["abc123"], dir.path().join("abc123.png"));
}
2 changes: 1 addition & 1 deletion crates/openhuman-core/src/agent/session_host/driver.rs
Original file line number Diff line number Diff line change
Expand Up @@ -146,7 +146,7 @@ impl SessionDriver<OpenHumanRunContext> for OpenHumanSessionDriver {
.filter_map(tinyagents_runtime::transcript::message_to_native_transcript)
.collect();
if crate::agent::multimodal::has_image_placeholders(&messages) {
messages = crate::agent::multimodal::rehydrate_image_placeholders(&messages);
messages = crate::agent::multimodal::rehydrate_image_placeholders(&messages).await;
}

// Runtime's `ToolSnapshot` remains the model declaration and final
Expand Down
7 changes: 4 additions & 3 deletions crates/openhuman-core/src/agent/session_store/mod.rs
Original file line number Diff line number Diff line change
Expand Up @@ -26,9 +26,10 @@ mod workspace;

pub use rebound::{rebind_all, ReboundSessionStores};
pub use workspace::{
current_documents_for, ledger_for, provider_for_workspace, release_workspace_defaults,
require_ledger, status_store_for, stores_for, stores_or_ephemeral, with_ledger,
with_ledger_async, WorkspaceSessionStores,
attachment_stash_for, current_documents_for, default_turn_states_for, ledger_for,
provider_for_workspace, release_workspace_defaults, require_ledger, status_store_for,
stores_for, stores_or_ephemeral, turn_states_for, with_ledger, with_ledger_async,
WorkspaceSessionStores,
};

static PROVIDER: LazyLock<RwLock<Option<Arc<dyn SessionStoreProvider>>>> =
Expand Down
Loading
Loading