diff --git a/src/openhuman/agent/harness/mod.rs b/src/openhuman/agent/harness/mod.rs index 3edab2f05..57c910b82 100644 --- a/src/openhuman/agent/harness/mod.rs +++ b/src/openhuman/agent/harness/mod.rs @@ -40,6 +40,7 @@ pub mod session; pub(crate) mod session_queue; pub(crate) mod spawn_depth_context; pub mod subagent_runner; +pub mod task_recency_context; mod token_budget; pub(crate) mod tool_filter; mod tool_loop; @@ -57,6 +58,7 @@ pub use model_vision_context::{current_model_vision, with_current_model_vision}; pub use sandbox_context::{current_sandbox_mode, with_current_sandbox_mode}; pub(crate) use spawn_depth_context::{current_spawn_depth, with_spawn_depth, MAX_SPAWN_DEPTH}; pub use subagent_runner::{run_subagent, SubagentRunError, SubagentRunOptions}; +pub use task_recency_context::{current_task_recency_window, with_task_recency_window}; pub use worktree_context::{current_action_dir_override, with_action_dir_override}; pub(crate) use instructions::build_tool_instructions_filtered; diff --git a/src/openhuman/agent/harness/task_recency_context.rs b/src/openhuman/agent/harness/task_recency_context.rs new file mode 100644 index 000000000..9980988f6 --- /dev/null +++ b/src/openhuman/agent/harness/task_recency_context.rs @@ -0,0 +1,106 @@ +//! Task-local carrier for a **task-recency window** so the Composio tool +//! surface can restrict task-fetch results to "recently created/changed" +//! without widening the [`crate::openhuman::tools::Tool`] trait signature. +//! +//! Sibling of [`super::sandbox_context`]: same task-local pattern, different +//! concept. `CURRENT_AGENT_SANDBOX_MODE` carries the calling agent's sandbox +//! mode; [`TASK_RECENCY_WINDOW`] carries an optional "only data newer than +//! `now - window`" hint that the `composio_execute` handler applies to a +//! curated set of task-fetch slugs (see `composio::task_window`). +//! +//! Why a task-local instead of a `Tool::execute` argument: the tool trait is +//! invoked from many call sites (CLI, JSON-RPC, tests, agent loops). A +//! task-local keeps the additive path scoped to the one agent runtime that +//! needs it — today only the `morning_briefing` cron agent installs a window. +//! +//! When the task-local isn't set (normal chat, CLI, JSON-RPC, unit tests), +//! [`current_task_recency_window`] returns `None` and the tool surface keeps +//! its default unbounded behavior. This is strictly additive: a user asking +//! "show all my Linear issues" in chat is never silently 24h-filtered. + +use std::time::Duration; + +tokio::task_local! { + /// Recency window installed for the currently-executing agent turn. + /// Scoped per turn by the cron agent runner; any tool executed inside + /// that turn can read it. Unset (→ `None`) everywhere else. + pub static TASK_RECENCY_WINDOW: Duration; +} + +/// Returns the active task-recency window, if the scope is installed. +/// +/// `None` outside [`with_task_recency_window`] — i.e. normal chat turns, +/// direct CLI/JSON-RPC tool dispatch, or unit tests calling a [`Tool`] +/// directly. Callers treat `None` as "no recency restriction". +pub fn current_task_recency_window() -> Option { + let window = TASK_RECENCY_WINDOW.try_with(|w| *w).ok(); + tracing::trace!( + has_window = window.is_some(), + window_secs = window.map(|w| w.as_secs()), + "[harness][task-window] read current window" + ); + window +} + +/// Run `future` with `window` installed as the current task-recency window. +/// +/// Intended call site is the cron agent runner, wrapped around the turn for +/// agents (today: `morning_briefing`) that should only see recently-touched +/// task data. The scope does not leak into detached tasks spawned inside +/// `future` — standard [`tokio::task_local!`] semantics. +pub async fn with_task_recency_window(window: Duration, future: F) -> R +where + F: std::future::Future, +{ + tracing::trace!( + window_secs = window.as_secs(), + "[harness][task-window] scope enter" + ); + let out = TASK_RECENCY_WINDOW.scope(window, future).await; + tracing::trace!("[harness][task-window] scope exit"); + out +} + +#[cfg(test)] +mod tests { + use super::*; + + #[tokio::test] + async fn current_window_returns_none_outside_scope() { + assert_eq!(current_task_recency_window(), None); + } + + #[tokio::test] + async fn with_window_installs_value() { + let observed = with_task_recency_window(Duration::from_secs(86_400), async { + current_task_recency_window() + }) + .await; + assert_eq!(observed, Some(Duration::from_secs(86_400))); + } + + #[tokio::test] + async fn with_window_does_not_leak_across_scopes() { + with_task_recency_window(Duration::from_secs(60), async { + assert_eq!(current_task_recency_window(), Some(Duration::from_secs(60))); + }) + .await; + assert_eq!(current_task_recency_window(), None); + } + + #[tokio::test] + async fn nested_scope_overrides_outer() { + with_task_recency_window(Duration::from_secs(60), async { + assert_eq!(current_task_recency_window(), Some(Duration::from_secs(60))); + with_task_recency_window(Duration::from_secs(120), async { + assert_eq!( + current_task_recency_window(), + Some(Duration::from_secs(120)) + ); + }) + .await; + assert_eq!(current_task_recency_window(), Some(Duration::from_secs(60))); + }) + .await; + } +} diff --git a/src/openhuman/agent_registry/agents/morning_briefing/prompt.md b/src/openhuman/agent_registry/agents/morning_briefing/prompt.md index c9493a177..61d48ee09 100644 --- a/src/openhuman/agent_registry/agents/morning_briefing/prompt.md +++ b/src/openhuman/agent_registry/agents/morning_briefing/prompt.md @@ -9,7 +9,7 @@ Prepare a morning briefing that helps the user start their day with clarity. Pul ## What to include (in priority order) 1. **Calendar** — Today's meetings, calls, and events. Lead times, conflicts, and gaps worth noting. -2. **Tasks & action items** — Open to-dos, deadlines due today, and anything overdue that needs attention. +2. **Tasks & action items** — To-dos **created or changed in the last 24h**, deadlines due today, and anything overdue that needs attention. The system already restricts task-tool results to that 24h window, so treat what `composio_execute` returns for a task manager as recent by construction — don't try to re-fetch the whole backlog. 3. **Important emails / messages** — Unread threads that look time-sensitive or are from key contacts. Don't list every newsletter. 4. **Crypto / market context** — If the user tracks markets, surface notable overnight moves, liquidation events, or governance votes closing today. Keep it to 2-3 bullets max. 5. **Recent memory** — What actually happened across the user's connected sources in the **last 24 hours** (conversations, threads, activity), plus any commitment now due (e.g. "you said you'd finish the proposal by Wednesday" — and today is Wednesday). diff --git a/src/openhuman/composio/mod.rs b/src/openhuman/composio/mod.rs index ef7c406cc..4baee0e4f 100644 --- a/src/openhuman/composio/mod.rs +++ b/src/openhuman/composio/mod.rs @@ -50,6 +50,7 @@ pub mod ops; pub mod periodic; pub mod providers; pub mod schemas; +pub mod task_window; pub mod tools; pub mod trigger_history; pub mod types; diff --git a/src/openhuman/composio/task_window.rs b/src/openhuman/composio/task_window.rs new file mode 100644 index 000000000..0780033f7 --- /dev/null +++ b/src/openhuman/composio/task_window.rs @@ -0,0 +1,323 @@ +//! Recency-window narrowing for Composio **task-fetch** actions. +//! +//! Background: the `morning_briefing` agent fetches the user's open tasks via +//! `composio_execute` against whichever task manager they connected. Left +//! unbounded those calls return the *entire* backlog; the brief only wants +//! what was created/changed in the last 24h. The cron runner installs a +//! [`crate::openhuman::agent::harness::current_task_recency_window`] for the +//! brief turn, and this module applies it inside the `composio_execute` +//! handler. +//! +//! Two layers, both gated on the task-local window being present AND the slug +//! appearing in [`spec_for`]: +//! +//! 1. **Best-effort server-side narrowing** ([`apply_window_args`]): inject an +//! ordering / `*_since` argument the provider understands, when the caller +//! didn't supply one. Caller-supplied values always win. This only reduces +//! payload size / improves ordering — correctness never depends on it. +//! 2. **Authoritative client-side post-filter** ([`filter_response`]): drop +//! rows whose timestamp predates `now - window`. This is the enforcement. +//! Mirrors the proven `sync_depth_days` floor in the native sync providers +//! (e.g. `memory_sync::composio::providers::linear::provider`). +//! +//! Scope is intentionally narrow: only slugs with a *verified* response shape +//! are listed. An unknown slug degrades to "no filtering" — never a crash and never a +//! silently-emptied result. An unrecognized envelope or an unparseable +//! timestamp is treated as "keep", so a wrong field-map degrades to no-op. +//! +//! `markdownFormatted` note: in backend mode the response often carries a +//! server-rendered markdown string that the tool surface prefers over `data`. +//! When the post-filter removes rows it clears `markdown_formatted` so the +//! agent reads the *filtered* JSON instead of the stale full-backlog markdown. + +use chrono::{DateTime, Utc}; +use serde_json::Value; + +use super::types::ComposioExecuteResponse; + +/// How a provider encodes a task timestamp, so the post-filter can compare it. +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +enum TsFormat { + /// RFC3339 / ISO-8601 string, e.g. `2026-06-21T10:00:00.000Z`. + Iso8601, + /// Unix epoch **milliseconds**, as a JSON string or number (ClickUp). + EpochMillis, +} + +/// Per-slug knowledge needed to narrow + filter a task-fetch response. +struct TaskWindowSpec { + /// Candidate key-paths to the items array inside `resp.data`. The first + /// path that resolves to a JSON array wins (providers differ in whether + /// they nest under `data`, `results`, `nodes`, …). An empty path means + /// `resp.data` is itself the array. + items_paths: &'static [&'static [&'static str]], + /// Timestamp fields on each item, checked in order. A row is **kept** if + /// *any* listed field parses to a time `>= floor`. A row whose fields are + /// all missing/unparseable is also kept (we never drop on ambiguity). + ts_fields: &'static [(&'static str, TsFormat)], + /// Optional static ordering arg to inject (key, value) when absent. + order_arg: Option<(&'static str, &'static str)>, +} + +/// Verified task-fetch slugs. Keep this list to slugs whose response shape is +/// confirmed (repo-proven or docs-verified). See module docs. +fn spec_for(slug: &str) -> Option { + match slug { + // Linear — repo-proven request (`orderBy:"updatedAt"`) and response + // shape. Linear returns GraphQL connections (`{issues:{nodes:[...]}}`) + // that Composio may re-wrap under `data` / `data.data`; the items and + // timestamp paths mirror `providers::linear::sync::{extract_issues, + // extract_issue_updated}` so we catch every nesting (else we'd hit the + // no-array pass-through and leave the backlog unfiltered). + "LINEAR_LIST_LINEAR_ISSUES" | "LINEAR_SEARCH_ISSUES" => Some(TaskWindowSpec { + items_paths: &[ + &["data", "issues", "nodes"], + &["data", "data", "nodes"], + &["data", "nodes"], + &["nodes"], + &["issues", "nodes"], + &["data", "results"], + &["results"], + &["data", "issues"], + &["issues"], + ], + ts_fields: &[ + ("updatedAt", TsFormat::Iso8601), + ("data.updatedAt", TsFormat::Iso8601), + ("updated_at", TsFormat::Iso8601), + ("data.updated_at", TsFormat::Iso8601), + ], + order_arg: Some(("orderBy", "updatedAt")), + }), + // ClickUp — repo-proven request (`order_by:"updated"`) and response + // shape, mirroring `providers::clickup::sync::{extract_tasks, + // extract_task_updated}` (envelope `data.tasks` / `tasks` / + // `data.data.tasks`; timestamp `date_updated` epoch-ms, possibly + // wrapped or camelCased by Composio). + "CLICKUP_GET_FILTERED_TEAM_TASKS" | "CLICKUP_GET_TASKS" => Some(TaskWindowSpec { + items_paths: &[ + &["data", "tasks"], + &["tasks"], + &["data", "data", "tasks"], + &["data", "results"], + &["results"], + ], + ts_fields: &[ + ("date_updated", TsFormat::EpochMillis), + ("data.date_updated", TsFormat::EpochMillis), + ("dateUpdated", TsFormat::EpochMillis), + ("data.dateUpdated", TsFormat::EpochMillis), + ], + order_arg: Some(("order_by", "updated")), + }), + // Notion — docs-verified: every page carries `last_edited_time` and + // `created_time`; query results live under `results` (Composio may + // re-wrap under `data` / `data.data`). + "NOTION_QUERY_DATABASE_WITH_FILTER" => Some(TaskWindowSpec { + items_paths: &[ + &["results"], + &["data", "results"], + &["data", "data", "results"], + ], + ts_fields: &[ + ("last_edited_time", TsFormat::Iso8601), + ("created_time", TsFormat::Iso8601), + ("data.last_edited_time", TsFormat::Iso8601), + ("data.created_time", TsFormat::Iso8601), + ], + order_arg: None, + }), + // Asana — docs-verified server-side `modified_since` (injected in + // apply_window_args). Items live under `data` (Composio may re-wrap as + // `data.data`); each row carries `modified_at` / `created_at`, possibly + // `data`-wrapped. + // CONFIRM-AT-RUNTIME: item timestamp field names via composio_list_tools. + "ASANA_GET_MULTIPLE_TASKS" => Some(TaskWindowSpec { + items_paths: &[&["data", "data"], &["data"]], + ts_fields: &[ + ("modified_at", TsFormat::Iso8601), + ("created_at", TsFormat::Iso8601), + ("data.modified_at", TsFormat::Iso8601), + ("data.created_at", TsFormat::Iso8601), + ], + order_arg: None, + }), + _ => None, + } +} + +/// Inject best-effort server-side narrowing args for a task-fetch slug, when +/// the task-recency window is active. Caller-supplied keys are never +/// overwritten. Unknown slugs pass through untouched. +pub(crate) fn apply_window_args( + slug: &str, + arguments: Option, + since: DateTime, +) -> Option { + let Some(spec) = spec_for(slug) else { + return arguments; + }; + + let mut value = arguments.unwrap_or_else(|| Value::Object(Default::default())); + let Some(map) = value.as_object_mut() else { + // Non-object payload (unexpected) — leave it for the backend to reject. + return Some(value); + }; + + if let Some((key, val)) = spec.order_arg { + map.entry(key.to_string()) + .or_insert_with(|| Value::String(val.to_string())); + } + + // Asana takes a dynamic `modified_since` rather than a static order arg. + if slug == "ASANA_GET_MULTIPLE_TASKS" { + map.entry("modified_since".to_string()) + .or_insert_with(|| Value::String(since.to_rfc3339())); + } + + tracing::debug!( + target: "composio", + slug, + since = %since.to_rfc3339(), + "[composio][task-window] applied server-side narrowing args" + ); + Some(value) +} + +/// Drop task rows older than `since` from a successful response. +/// +/// No-op when: the response failed, the slug isn't recognized, the items +/// array can't be located, or nothing was actually removed. When rows *are* +/// removed, `markdown_formatted` is cleared so the agent consumes the filtered +/// `data` rather than stale server-rendered markdown. +pub(crate) fn filter_response( + slug: &str, + mut resp: ComposioExecuteResponse, + since: DateTime, +) -> ComposioExecuteResponse { + if !resp.successful { + return resp; + } + let Some(spec) = spec_for(slug) else { + return resp; + }; + + let Some(path) = spec + .items_paths + .iter() + .find(|p| array_at(&resp.data, p).is_some()) + .copied() + else { + tracing::debug!( + target: "composio", + slug, + "[composio][task-window] no items array located; pass-through" + ); + return resp; + }; + + // Safe: `find` above guaranteed this path resolves to an array. + let items = array_at(&resp.data, path).expect("path resolved above"); + let total = items.len(); + let kept: Vec = items + .iter() + .filter(|item| keep_item(item, spec.ts_fields, since)) + .cloned() + .collect(); + let removed = total - kept.len(); + + if removed == 0 { + return resp; + } + + if let Some(slot) = array_at_mut(&mut resp.data, path) { + *slot = kept; + } + // Stale server-rendered markdown reflects the full backlog — force the + // tool surface to fall back to the filtered JSON envelope. + resp.markdown_formatted = None; + + tracing::debug!( + target: "composio", + slug, + kept = total - removed, + total, + since = %since.to_rfc3339(), + "[composio][task-window] post-filtered task results to recency window" + ); + resp +} + +/// Keep a row if any of its timestamp fields parses to `>= floor`. A row with +/// no parseable timestamp field is kept (never dropped on ambiguity). +/// +/// Field names may be dotted (`data.updatedAt`) to reach Composio-wrapped rows +/// — mirrors `providers::pick_str`, which the native sync extractors use for +/// exactly these `data.`-nested envelopes. +fn keep_item(item: &Value, ts_fields: &[(&str, TsFormat)], floor: DateTime) -> bool { + let mut saw_timestamp = false; + for (field, fmt) in ts_fields { + if let Some(raw) = field_at(item, field) { + if let Some(ts) = parse_ts(raw, *fmt) { + saw_timestamp = true; + if ts >= floor { + return true; + } + } + } + } + // No usable timestamp at all → conservatively keep. + !saw_timestamp +} + +/// Resolve a possibly-dotted field path (`updatedAt`, `data.updatedAt`) within +/// a single item object. Each segment is an object key. +fn field_at<'a>(item: &'a Value, dotted: &str) -> Option<&'a Value> { + let mut cur = item; + for seg in dotted.split('.') { + cur = cur.get(seg)?; + } + Some(cur) +} + +/// Parse a JSON timestamp value in the given format. Returns `None` on any +/// shape we don't recognize so the caller treats it as ambiguous (keep). +fn parse_ts(raw: &Value, fmt: TsFormat) -> Option> { + match fmt { + TsFormat::Iso8601 => raw + .as_str() + .and_then(|s| DateTime::parse_from_rfc3339(s.trim()).ok()) + .map(|dt| dt.with_timezone(&Utc)), + TsFormat::EpochMillis => { + let ms = match raw { + Value::String(s) => s.trim().parse::().ok(), + Value::Number(n) => n.as_i64(), + _ => None, + }?; + DateTime::from_timestamp_millis(ms) + } + } +} + +/// Resolve `path` (a slice of object keys) to an array within `root`. +/// An empty path means `root` itself must be the array. +fn array_at<'a>(root: &'a Value, path: &[&str]) -> Option<&'a Vec> { + let mut cur = root; + for key in path { + cur = cur.get(*key)?; + } + cur.as_array() +} + +/// Mutable counterpart to [`array_at`]. +fn array_at_mut<'a>(root: &'a mut Value, path: &[&str]) -> Option<&'a mut Vec> { + let mut cur = root; + for key in path { + cur = cur.get_mut(*key)?; + } + cur.as_array_mut() +} + +#[cfg(test)] +#[path = "task_window_tests.rs"] +mod tests; diff --git a/src/openhuman/composio/task_window_tests.rs b/src/openhuman/composio/task_window_tests.rs new file mode 100644 index 000000000..c21a913df --- /dev/null +++ b/src/openhuman/composio/task_window_tests.rs @@ -0,0 +1,203 @@ +//! Tests for the task-recency window narrowing + post-filter. + +use super::*; +use chrono::TimeZone; +use serde_json::json; + +fn floor() -> DateTime { + // Fixed "now - 24h" floor for deterministic comparisons. + Utc.with_ymd_and_hms(2026, 6, 20, 9, 0, 0).unwrap() +} + +fn ok_resp(data: Value) -> ComposioExecuteResponse { + ComposioExecuteResponse { + data, + successful: true, + error: None, + cost_usd: 0.0, + markdown_formatted: None, + } +} + +// ── post-filter: ClickUp (epoch-millis string) ────────────────────── + +#[test] +fn clickup_drops_rows_older_than_floor() { + let old_ms = (floor().timestamp_millis() - 1).to_string(); + let new_ms = (floor().timestamp_millis() + 60_000).to_string(); + let resp = ok_resp(json!({ + "data": { "tasks": [ + { "id": "old", "date_updated": old_ms }, + { "id": "new", "date_updated": new_ms }, + ]} + })); + let out = filter_response("CLICKUP_GET_FILTERED_TEAM_TASKS", resp, floor()); + let tasks = out.data["data"]["tasks"].as_array().unwrap(); + assert_eq!(tasks.len(), 1); + assert_eq!(tasks[0]["id"], "new"); +} + +#[test] +fn clickup_keeps_row_exactly_on_floor() { + let on_ms = floor().timestamp_millis().to_string(); + let resp = ok_resp(json!({ "tasks": [{ "id": "edge", "date_updated": on_ms }] })); + let out = filter_response("CLICKUP_GET_FILTERED_TEAM_TASKS", resp, floor()); + assert_eq!(out.data["tasks"].as_array().unwrap().len(), 1); +} + +// ── post-filter: Linear (ISO-8601, GraphQL `nodes` envelope) ───────── + +#[test] +fn linear_filters_iso_timestamps_under_nodes() { + let resp = ok_resp(json!({ + "nodes": [ + { "id": "old", "updatedAt": "2026-06-19T09:00:00.000Z" }, + { "id": "new", "updatedAt": "2026-06-20T12:00:00.000Z" }, + ] + })); + let out = filter_response("LINEAR_LIST_LINEAR_ISSUES", resp, floor()); + let nodes = out.data["nodes"].as_array().unwrap(); + assert_eq!(nodes.len(), 1); + assert_eq!(nodes[0]["id"], "new"); +} + +#[test] +fn linear_filters_graphql_connection_under_data_issues_nodes() { + // Real Linear/Composio shape: { data: { issues: { nodes: [...] } } }. + // Regression for the no-array pass-through that left the backlog unfiltered. + let resp = ok_resp(json!({ + "data": { "issues": { "nodes": [ + { "id": "old", "updatedAt": "2026-06-19T09:00:00.000Z" }, + { "id": "new", "updatedAt": "2026-06-20T12:00:00.000Z" }, + ]}} + })); + let out = filter_response("LINEAR_LIST_LINEAR_ISSUES", resp, floor()); + let nodes = out.data["data"]["issues"]["nodes"].as_array().unwrap(); + assert_eq!(nodes.len(), 1); + assert_eq!(nodes[0]["id"], "new"); +} + +#[test] +fn filters_data_wrapped_row_timestamps() { + // Composio sometimes wraps each row under `data` — the timestamp must + // still be read (else stale rows survive as "no timestamp → keep"). + let resp = ok_resp(json!({ + "tasks": [ + { "data": { "id": "old", "date_updated": "0" } }, + { "data": { "id": "new", + "date_updated": (floor().timestamp_millis()+1).to_string() } }, + ] + })); + let out = filter_response("CLICKUP_GET_FILTERED_TEAM_TASKS", resp, floor()); + let tasks = out.data["tasks"].as_array().unwrap(); + assert_eq!(tasks.len(), 1); + assert_eq!(tasks[0]["data"]["id"], "new"); +} + +// ── post-filter: Notion (keep if EITHER timestamp is fresh) ────────── + +#[test] +fn notion_keeps_page_when_either_timestamp_fresh() { + let resp = ok_resp(json!({ + "results": [ + // created long ago but recently edited → keep + { "id": "edited", "created_time": "2026-01-01T00:00:00.000Z", + "last_edited_time": "2026-06-20T10:00:00.000Z" }, + // both stale → drop + { "id": "stale", "created_time": "2026-01-01T00:00:00.000Z", + "last_edited_time": "2026-01-02T00:00:00.000Z" }, + ] + })); + let out = filter_response("NOTION_QUERY_DATABASE_WITH_FILTER", resp, floor()); + let results = out.data["results"].as_array().unwrap(); + assert_eq!(results.len(), 1); + assert_eq!(results[0]["id"], "edited"); +} + +// ── conservative behaviors ────────────────────────────────────────── + +#[test] +fn rows_without_timestamp_are_kept() { + let resp = ok_resp(json!({ "tasks": [{ "id": "no_ts" }] })); + let out = filter_response("CLICKUP_GET_FILTERED_TEAM_TASKS", resp, floor()); + assert_eq!(out.data["tasks"].as_array().unwrap().len(), 1); +} + +#[test] +fn unparseable_timestamp_is_kept() { + let resp = ok_resp(json!({ "tasks": [{ "id": "junk", "date_updated": "not-a-number" }] })); + let out = filter_response("CLICKUP_GET_FILTERED_TEAM_TASKS", resp, floor()); + assert_eq!(out.data["tasks"].as_array().unwrap().len(), 1); +} + +#[test] +fn unknown_slug_is_untouched() { + let resp = ok_resp(json!({ "tasks": [{ "id": "x", "date_updated": "0" }] })); + let out = filter_response("GMAIL_FETCH_EMAILS", resp, floor()); + assert_eq!(out.data["tasks"].as_array().unwrap().len(), 1); +} + +#[test] +fn failed_response_is_untouched() { + let old_ms = "0".to_string(); + let mut resp = ok_resp(json!({ "tasks": [{ "id": "old", "date_updated": old_ms }] })); + resp.successful = false; + let out = filter_response("CLICKUP_GET_FILTERED_TEAM_TASKS", resp, floor()); + assert_eq!(out.data["tasks"].as_array().unwrap().len(), 1); +} + +#[test] +fn missing_items_array_is_pass_through() { + let resp = ok_resp(json!({ "unexpected": "shape" })); + let out = filter_response("CLICKUP_GET_FILTERED_TEAM_TASKS", resp, floor()); + assert_eq!(out.data["unexpected"], "shape"); +} + +// ── markdown_formatted invalidation ───────────────────────────────── + +#[test] +fn markdown_cleared_only_when_rows_removed() { + // Rows removed → markdown dropped so agent reads filtered JSON. + let mut resp = ok_resp(json!({ "tasks": [ + { "id": "old", "date_updated": "0" }, + { "id": "new", "date_updated": (floor().timestamp_millis()+1).to_string() }, + ]})); + resp.markdown_formatted = Some("| old | new | full backlog |".into()); + let out = filter_response("CLICKUP_GET_FILTERED_TEAM_TASKS", resp, floor()); + assert!(out.markdown_formatted.is_none()); + + // Nothing removed → markdown preserved. + let mut resp2 = ok_resp(json!({ "tasks": [ + { "id": "new", "date_updated": (floor().timestamp_millis()+1).to_string() }, + ]})); + resp2.markdown_formatted = Some("kept".into()); + let out2 = filter_response("CLICKUP_GET_FILTERED_TEAM_TASKS", resp2, floor()); + assert_eq!(out2.markdown_formatted.as_deref(), Some("kept")); +} + +// ── apply_window_args ─────────────────────────────────────────────── + +#[test] +fn injects_order_arg_when_absent() { + let out = apply_window_args("LINEAR_LIST_LINEAR_ISSUES", None, floor()).unwrap(); + assert_eq!(out["orderBy"], "updatedAt"); +} + +#[test] +fn caller_supplied_order_arg_wins() { + let args = json!({ "order_by": "created" }); + let out = apply_window_args("CLICKUP_GET_FILTERED_TEAM_TASKS", Some(args), floor()).unwrap(); + assert_eq!(out["order_by"], "created"); +} + +#[test] +fn asana_gets_modified_since_injected() { + let out = apply_window_args("ASANA_GET_MULTIPLE_TASKS", Some(json!({})), floor()).unwrap(); + assert_eq!(out["modified_since"], floor().to_rfc3339()); +} + +#[test] +fn apply_window_args_noop_for_unknown_slug() { + let out = apply_window_args("GMAIL_FETCH_EMAILS", Some(json!({ "a": 1 })), floor()); + assert_eq!(out.unwrap(), json!({ "a": 1 })); +} diff --git a/src/openhuman/composio/tools.rs b/src/openhuman/composio/tools.rs index 9e0f63163..ed5dbe56b 100644 --- a/src/openhuman/composio/tools.rs +++ b/src/openhuman/composio/tools.rs @@ -31,6 +31,7 @@ use async_trait::async_trait; use serde_json::{json, Value}; use crate::openhuman::agent::harness::current_sandbox_mode; +use crate::openhuman::agent::harness::current_task_recency_window; use crate::openhuman::agent::harness::definition::SandboxMode; use crate::openhuman::config::rpc as config_rpc; use crate::openhuman::config::Config; @@ -1078,6 +1079,20 @@ impl Tool for ComposioExecuteTool { let arguments = super::googlecalendar_args::apply_calendar_query_defaults(&tool, arguments, &iana); + // Task-recency window (morning briefing): when the calling agent + // installed a window, inject best-effort server-side narrowing for + // curated task-fetch slugs. No-op for every other slug and for + // normal chat / CLI / JSON-RPC (window unset → None). The + // authoritative enforcement is the post-filter on the response below. + let task_window_since = current_task_recency_window().map(|w| { + chrono::Utc::now() + - chrono::Duration::from_std(w).unwrap_or_else(|_| chrono::Duration::zero()) + }); + let arguments = match task_window_since { + Some(since) => super::task_window::apply_window_args(&tool, arguments, since), + None => arguments, + }; + // Resolve the client through the mode-aware factory on every // call so a direct-mode toggle takes effect immediately // (#1710). The pre-baked-client variant of this code routed all @@ -1122,6 +1137,14 @@ impl Tool for ComposioExecuteTool { let elapsed_ms = started.elapsed().as_millis() as u64; match res { Ok(resp) => { + // Authoritative task-recency enforcement: drop task rows older + // than the window. No-op unless a window is installed AND the + // slug is a curated task-fetch action. Runs before the + // markdown/JSON body decision so the agent reads filtered data. + let resp = match task_window_since { + Some(since) => super::task_window::filter_response(&tool, resp, since), + None => resp, + }; tracing::info!( tool = %tool, successful = resp.successful, diff --git a/src/openhuman/cron/scheduler.rs b/src/openhuman/cron/scheduler.rs index a76f4d941..3856d32a1 100644 --- a/src/openhuman/cron/scheduler.rs +++ b/src/openhuman/cron/scheduler.rs @@ -20,6 +20,10 @@ const SHELL_JOB_TIMEOUT_SECS: u64 = 120; const AGENT_JOB_USER_FAILURE_MESSAGE: &str = "Something went wrong. Please try again.\nThis error has been reported. You can also report it on Discord.\nReport on Discord"; const MORNING_BRIEFING_AGENT_ID: &str = "morning_briefing"; const MORNING_BRIEFING_FAILURE_NOTIFICATION: &str = "Morning briefing could not run. Check your AI provider, API key, and connected apps, then run it again from Settings > Cron Jobs."; +/// Recency window the morning briefing installs around its turn so Composio +/// task-fetch tools only surface tasks created/changed in the last day. Read +/// by the `composio_execute` handler via `current_task_recency_window`. +const MORNING_BRIEFING_TASK_RECENCY_SECS: u64 = 24 * 60 * 60; /// Map a typed [`AgentError`] to a canned, user-facing message for cron-job /// failure notifications. @@ -578,11 +582,32 @@ async fn run_agent_job(config: &Config, job: &CronJob) -> (bool, String, Option< source: crate::openhuman::agent::turn_origin::TrustedAutomationSource::Cron, }; - crate::openhuman::agent::turn_origin::with_origin( + let turn = crate::openhuman::agent::turn_origin::with_origin( origin, agent.run_single(&prefixed_prompt), - ) - .await + ); + // Morning briefing only: install a 24h task-recency window + // so Composio task-fetch tools (Linear/ClickUp/Notion/Asana) + // surface only recently created/changed tasks. Other cron + // agents and all chat turns leave the window unset. + if is_morning_briefing_job(job) { + tracing::debug!( + job_id = %job.id, + recency_window_secs = MORNING_BRIEFING_TASK_RECENCY_SECS, + "[cron] applying morning-briefing task recency window" + ); + crate::openhuman::agent::harness::with_task_recency_window( + std::time::Duration::from_secs(MORNING_BRIEFING_TASK_RECENCY_SECS), + turn, + ) + .await + } else { + tracing::trace!( + job_id = %job.id, + "[cron] task recency window not applied for this job" + ); + turn.await + } } Err(e) => Err(e), }