mirror of
https://github.com/tinyhumansai/openhuman.git
synced 2026-07-27 21:08:00 +00:00
feat(briefing): scope morning-brief task fetch to a 24h recency window (#3872)
Co-authored-by: Claude <noreply@anthropic.com> Co-authored-by: Steven Enamakel <enamakel@tinyhumans.ai>
This commit is contained in:
sanil-23
co-authored by
Claude
Steven Enamakel
Claude
Steven Enamakel
parent
e159c7a1cb
commit
3e0244670c
@@ -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;
|
||||
|
||||
@@ -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<Duration> {
|
||||
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<F, R>(window: Duration, future: F) -> R
|
||||
where
|
||||
F: std::future::Future<Output = R>,
|
||||
{
|
||||
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;
|
||||
}
|
||||
}
|
||||
@@ -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).
|
||||
|
||||
@@ -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;
|
||||
|
||||
@@ -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<TaskWindowSpec> {
|
||||
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<Value>,
|
||||
since: DateTime<Utc>,
|
||||
) -> Option<Value> {
|
||||
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<Utc>,
|
||||
) -> 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<Value> = 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<Utc>) -> 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<DateTime<Utc>> {
|
||||
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::<i64>().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<Value>> {
|
||||
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<Value>> {
|
||||
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;
|
||||
@@ -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<Utc> {
|
||||
// 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 }));
|
||||
}
|
||||
@@ -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,
|
||||
|
||||
@@ -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.\n<openhuman-link path=\"community/discord-report\">Report on Discord</openhuman-link>";
|
||||
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),
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user