scriptoriumadmin/ReMe
1
0
Fork 0
mirror of https://github.com/agentscope-ai/ReMe.git synced 2026-10-23 05:35:30 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 9

feat(proactive): accept host delivery receipts

Browse source 
The refresh chain decides what is worth proposing but cannot know what the
host actually surfaced. Let a host agent report an agenda item back, so the
next round withholds it instead of proposing the same thing twice.

- proactive_feedback job: zero-LLM intake, idempotent upsert keyed on
  (topic_id, external_ref) and falling back to (topic_id, agenda_date,
  channel); status only moves forward out of a terminal value, and a failed
  report never blocks a retry
- truth source: optional ``deliveries`` section in daily/_proactive.yaml,
  trimmed by retention_days then max_deliveries, and salvaged entry by entry
  with the rest of the file
- reader side: proactive_read withholds agenda items inside
  delivery_cooldown_days and returns delivered_ids so a caller can see the
  suppression instead of guessing; include_delivered opts out
- agenda side: a reported candidate leaves the set before reminders are
  pinned and before the editor call, so the cooldown cannot be argued away
  by narrative coherence, and it is suppressed with an auditable
  "already delivered ..." reason
- tests: receipt intake, reader-side withholding and agenda-side suppression,
  each paired with an empty-deliveries no-change case, plus a round-trip of
  the report shape a consumer actually sends

Reporting stays optional: a host that never calls proactive_feedback keeps an
empty deliveries section and sees no behaviour change.
This commit is contained in:
imrewce 2026-10-12 15:36:56 +08:00
parent c0bad3d30e
commit a0f66c1779
10 changed files with 1124 additions and 25 deletions
Show stats Download patch file Download diff file Expand all files Collapse all files

33
reme/config/default.yaml
View file

@ -127,6 +127,7 @@ jobs:
max_reminder_cards: 8
- backend: proactive_agenda_step
min_discovery_slots: 2
redelivery_cooldown_days: 3
- backend: proactive_finish_step
file_catalog: proactive
@ -169,6 +170,7 @@ jobs:
max_reminder_cards: 8
- backend: proactive_agenda_step
min_discovery_slots: 2
redelivery_cooldown_days: 3
- backend: proactive_finish_step
file_catalog: proactive
@ -307,9 +309,40 @@ jobs:
type: number
description: "minimum topic confidence to return (default 0.4 sits safely below the 0.5 fallback; v1 topics fall back to 0.5)"
default: 0.4
include_delivered:
type: boolean
description: "also return agenda items the host already reported as delivered (default withholds them)"
default: false
delivery_cooldown_days:
type: integer
description: "days an already-delivered agenda item stays withheld; 0 withholds only same-day deliveries"
default: 3
steps:
- backend: proactive_step
min_confidence: 0.4
delivery_cooldown_days: 3
proactive_feedback:
backend: base
description: "Record which proactive agenda items a host agent already delivered or executed, so they are not repeated."
parameters:
type: object
properties:
date:
type: string
description: "YYYY-MM-DD the report belongs to; defaults to today in the application's timezone"
default: ""
items:
type: array
description: "delivery reports: [{topic_id, status, channel, consumer, outcome, external_ref, agenda_date}]"
items:
type: object
required:
- items
steps:
- backend: proactive_feedback_step
retention_days: 180
max_deliveries: 500
version:
backend: base

2
reme/schema/__init__.py
View file

@ -8,6 +8,7 @@ from .dream import (
IntegrateOutcome,
)
from .proactive import (
ProactiveDelivery,
ProactiveResult,
ProactiveSchedule,
ProactiveState,
@ -42,6 +43,7 @@ __all__ = [
"GraphSnapshotNode",
"IntegrateOutcome",
"JobConfig",
"ProactiveDelivery",
"ProactiveResult",
"ProactiveSchedule",
"ProactiveState",

80
reme/schema/proactive.py
View file

@ -26,6 +26,22 @@ SCHEDULE_STATUSES = ("open", "done", "cancelled")
RECURRENCES = ("", "daily", "weekly", "monthly", "yearly")
REMINDER_TRIGGERS = ("overdue", "due_today", "due_soon", "conflict")
# Delivery feedback (``daily/_proactive.yaml: deliveries``) vocabularies.
# A host agent reports back which agenda items it already surfaced or ran, so
# the reader and the agenda step can stop repeating them. Reporting is
# optional: a framework that never calls ``proactive_feedback`` leaves the
# section empty and the whole chain behaves exactly as before.
DELIVERY_STATUSES = ("shown", "executed", "dismissed", "failed")
DELIVERY_CHANNELS = ("in_conversation", "inbox", "notification", "other")
DEFAULT_DELIVERY_STATUS = "shown"
DEFAULT_DELIVERY_CHANNEL = "other"
# Statuses that retire an agenda item for the whole cooldown window. ``failed``
# is deliberately absent: a failed run should be retryable on the next round.
TERMINAL_DELIVERY_STATUSES = ("executed", "dismissed")
# ``outcome`` is a host-written summary; cap it so a chatty consumer cannot
# inflate the truth source without bound.
MAX_DELIVERY_OUTCOME_CHARS = 500
# Keys inside the untyped ``history`` audit trail that hold a date/datetime.
HISTORY_WHEN_KEYS = ("at", "from", "to")
@ -188,6 +204,57 @@ class ProactiveSchedule(BaseModel):
return cleaned
class ProactiveDelivery(BaseModel):
"""One host-side report that an agenda item was surfaced or executed.
The identity of a record is ``(topic_id, external_ref)`` when the host
supplies a reference (an inbox event id, a session id), and
``(topic_id, agenda_date, channel)`` otherwise, so repeated reports of the
same delivery update in place instead of piling up. ``status`` only moves
forward: ``shown`` may become ``executed``/``dismissed``/``failed``, but a
terminal record is never downgraded back to ``shown``.
"""
model_config = ConfigDict(extra="ignore")
topic_id: str = ""
agenda_date: str = ""
delivered_at: str = ""
channel: str = DEFAULT_DELIVERY_CHANNEL
consumer: str = ""
status: str = DEFAULT_DELIVERY_STATUS
outcome: str = ""
external_ref: str = ""
@field_validator("status", mode="before")
@classmethod
def _fallback_status(cls, value):
text = str(value or "").strip().lower()
return text if text in DELIVERY_STATUSES else DEFAULT_DELIVERY_STATUS
@field_validator("channel", mode="before")
@classmethod
def _fallback_channel(cls, value):
text = str(value or "").strip().lower()
return text if text in DELIVERY_CHANNELS else DEFAULT_DELIVERY_CHANNEL
@field_validator("agenda_date", "delivered_at", mode="before")
@classmethod
def _iso_when(cls, value):
return iso_text(value)
@field_validator("outcome", mode="before")
@classmethod
def _truncate_outcome(cls, value):
return str(value or "")[:MAX_DELIVERY_OUTCOME_CHARS]
def identity(self) -> tuple[str, str]:
"""Idempotency key: explicit host reference wins over the natural key."""
if self.external_ref:
return (self.topic_id, f"ref:{self.external_ref}")
return (self.topic_id, f"day:{self.agenda_date}|{self.channel}")
class ProactiveState(BaseModel):
"""Chain-shared proactive context state (``context['proactive']``).
@ -256,6 +323,8 @@ class ProactiveState(BaseModel):
scenario_cards: list[dict] = Field(default_factory=list)
agenda: list[dict] = Field(default_factory=list)
suppressed: list[dict] = Field(default_factory=list)
# Candidates withheld because the host already reported them delivered.
suppressed_delivered: int = 0
plan_llm_calls: int = 0
push: bool = False
file_skip_reason: str = ""
@ -263,6 +332,11 @@ class ProactiveState(BaseModel):
interests_written: bool = False
checkpoint_paths: list[str] = Field(default_factory=list)
duration_ms: int = 0
# Delivery-feedback round counters (metadata/log only).
deliveries_in: int = 0
deliveries_added: int = 0
deliveries_updated: int = 0
deliveries_dropped: int = 0
class ProactiveStateFile(BaseModel):
@ -272,6 +346,8 @@ class ProactiveStateFile(BaseModel):
keep its original age anchor (F2.4 reopen channel).
``schedules`` is the long-lived reminder registry; it is optional so
files written before the reminder branch existed still load.
``deliveries`` is the optional host-feedback log; it is likewise absent
from files written before delivery reporting existed.
"""
model_config = ConfigDict(extra="ignore")
@ -280,6 +356,7 @@ class ProactiveStateFile(BaseModel):
open_topics: list[ProactiveTopic] = Field(default_factory=list)
resolved: list[dict] = Field(default_factory=list)
schedules: list[ProactiveSchedule] = Field(default_factory=list)
deliveries: list[ProactiveDelivery] = Field(default_factory=list)
class ProactiveResult(BaseModel):
@ -295,3 +372,6 @@ class ProactiveResult(BaseModel):
push: bool | None = None
generated_at: str = ""
agenda: list[dict] = Field(default_factory=list)
# Agenda topic_ids withheld because the host already reported them as
# delivered; kept so a caller can see the suppression instead of guessing.
delivered_ids: list[str] = Field(default_factory=list)

2
reme/steps/evolve/proactive/__init__.py
View file

@ -2,6 +2,7 @@
from .agenda import ProactiveAgendaStep
from .extract import ProactiveExtractStep
from .feedback import ProactiveFeedbackStep
from .finish import ProactiveFinishStep
from .plan import ProactivePlanStep
from .proactive import ProactiveStep
@ -10,6 +11,7 @@ from .topics import ProactiveTopicsStep
__all__ = [
"ProactiveAgendaStep",
"ProactiveExtractStep",
"ProactiveFeedbackStep",
"ProactiveFinishStep",
"ProactivePlanStep",
"ProactiveStep",

53
reme/steps/evolve/proactive/agenda.py
View file

@ -19,6 +19,14 @@ Reminder priority is enforced here, not negotiated: every reminder card is
pinned to the front of the agenda (nearest due date first) before the editor
LLM sees anything, and only the slots the reminders leave are up for
selection. Pinned items are never suppressed and never counted as "silence".
Delivery feedback is enforced even earlier: a candidate the host already
reported as surfaced/executed inside ``redelivery_cooldown_days`` leaves the
candidate set before pinning and before the editor call, with an auditable
``already delivered ...`` suppression reason. This is the write-side twin of
the reader's filter, so a proposal cannot reappear tomorrow merely because
the host reported late. Hosts that never report keep an empty ``deliveries``
section and see no change.
"""
import asyncio
@ -34,10 +42,12 @@ from .utils import (
current_now,
interests_path_for,
load_personal_profile_block,
load_state,
norm_path,
parse_agenda_reply,
render_interests,
resolve_agent_wrapper,
suppressed_delivery_ids,
write_interests_if_changed,
)
@ -70,6 +80,15 @@ def _pin_reason(candidate: dict, day: str = "") -> str:
return f"{base} (due {due})" if due else base
def _delivered_reason(record) -> str:
"""Suppression reason naming when, where and how the item was already used."""
if record is None:
return "already delivered (within the redelivery cooldown)"
when = str(record.agenda_date or "") or str(record.delivered_at or "")[:10]
where = f" via {record.channel}" if record.channel else ""
return f"already delivered on {when}{where} ({record.status})"
@R.register("proactive_agenda_step")
class ProactiveAgendaStep(BaseStep):
"""Generate today's proactive agenda and render the enriched interests.yaml.
@ -86,6 +105,7 @@ class ProactiveAgendaStep(BaseStep):
llm_timeout_seconds: float = 120,
profile_rel_path: str = "profile.md",
profile_max_chars: int = 2500,
redelivery_cooldown_days: int = 3,
skip_key: str = "proactive_skip",
**kwargs,
):
@ -95,6 +115,7 @@ class ProactiveAgendaStep(BaseStep):
self.llm_timeout_seconds = float(llm_timeout_seconds)
self.profile_rel_path = str(profile_rel_path or "")
self.profile_max_chars = max(int(profile_max_chars), 0)
self.redelivery_cooldown_days = max(int(redelivery_cooldown_days), 0)
self.skip_key = skip_key
async def execute(self):
@ -124,6 +145,16 @@ class ProactiveAgendaStep(BaseStep):
cards_by_id = {str(c.get("topic_id") or ""): c for c in state.scenario_cards if isinstance(c, dict)}
carded = [c for c in candidates if str(c.get("id")) in cards_by_id]
over_budget = [c for c in candidates if str(c.get("id")) not in cards_by_id]
# Delivery feedback outranks every other selection rule: drop what the
# host already surfaced before pinning reminders or calling the editor,
# so the cooldown cannot be argued away by narrative coherence.
daily = state.daily_dir or daily_dir(self)
blocked = self._suppressed_deliveries(ws, daily, state.date)
delivered = [c for c in candidates if str(c.get("id")) in blocked]
if delivered:
carded = [c for c in carded if str(c.get("id")) not in blocked]
over_budget = [c for c in over_budget if str(c.get("id")) not in blocked]
state.suppressed_delivered = len(delivered)
# Reminders are pinned: every reminder - a due schedule or a freshly
# found one - always reaches today's agenda and is never traded away for
# narrative coherence. What the editor LLM gets
@ -139,7 +170,7 @@ class ProactiveAgendaStep(BaseStep):
slots = max(self.max_agenda_items - len(pinned_ids), self.min_discovery_slots)
self.logger.info(
f"[{self.name}] start candidates={len(candidates)} carded={len(carded)} "
f"over_budget={len(over_budget)} "
f"over_budget={len(over_budget)} already_delivered={len(delivered)} "
f"pinned_reminders={len(pinned_ids)} slots={slots} "
f"max_agenda_items={self.max_agenda_items}",
)
@ -171,6 +202,15 @@ class ProactiveAgendaStep(BaseStep):
agenda_ids = pinned_ids + [cid for cid in selected_ids if cid not in pinned_ids]
suppressed: list[dict] = []
for candidate in delivered:
record = blocked.get(str(candidate.get("id")))
suppressed.append(
{
"topic_id": str(candidate.get("id")),
"title": str(candidate.get("title") or ""),
"reason": _delivered_reason(record),
},
)
for candidate in over_budget:
is_reminder = str(candidate.get("kind")) == "reminder"
suppressed.append(
@ -241,8 +281,8 @@ class ProactiveAgendaStep(BaseStep):
state.duration_ms = state.duration_ms + int((time.monotonic() - started) * 1000)
self._store(state)
answer = (
f"Agenda: {len(agenda)} item(s), suppressed={len(suppressed)}, push={state.push}, "
f"written={written} to {rel_path}"
f"Agenda: {len(agenda)} item(s), suppressed={len(suppressed)} "
f"(already delivered={len(delivered)}), push={state.push}, written={written} to {rel_path}"
)
self.context.response.success = True
self.context.response.answer = answer
@ -339,6 +379,13 @@ class ProactiveAgendaStep(BaseStep):
suppressed_reasons = {cid: "capacity (deterministic fallback)" for cid in carded_ids if cid not in agenda_ids}
return agenda_ids, order_reasons, suppressed_reasons
def _suppressed_deliveries(self, ws, daily: str, day: str) -> dict:
"""topic_id -> blocking delivery record reported by the host."""
state_file, _needs_bootstrap = load_state(ws, daily)
if not state_file.deliveries:
return {}
return suppressed_delivery_ids(state_file, day, self.redelivery_cooldown_days)
def _profile_block(self, ws) -> str:
"""Digest-personal profile first, legacy single profile file as fallback."""
digest_dir = str(self.config_value("digest_dir"))

106
reme/steps/evolve/proactive/feedback.py Normal file
View file

@ -0,0 +1,106 @@
"""Proactive delivery feedback step: record what a host agent already surfaced.
The refresh chain decides what is worth raising; it cannot know what the host
actually raised. This step closes that loop by accepting a small report from
the consumer and appending it to the ``deliveries`` section of the truth source
``daily/_proactive.yaml``. Two readers benefit:
- ``proactive_step`` withholds agenda items still inside the cooldown window,
so the same proposal is not surfaced twice;
- ``proactive_agenda_step`` suppresses those candidates deterministically and
records an auditable reason instead of re-editing them into today's agenda.
Reporting is optional. A framework that never calls this step leaves
``deliveries`` empty and both readers behave exactly as they did before, which
keeps the contract additive rather than a behavioural change for every host.
The step is deliberately narrow: it never calls an LLM, never touches
``open_topics``/``resolved``/``schedules``, never rewrites ``interests.yaml``
and never advances a file catalog. It only appends feedback.
"""
import time
from ...base_step import BaseStep
from ....components import R
from ..dream.utils import daily_dir, today, workspace_dir
from .utils import (
current_now,
load_state,
save_state,
suppressed_delivery_ids,
trim_deliveries,
upsert_deliveries,
)
@R.register("proactive_feedback_step")
class ProactiveFeedbackStep(BaseStep):
"""Upsert host-reported agenda deliveries into the truth source.
Records are idempotent on ``(topic_id, external_ref)`` when the host
supplies a reference and on ``(topic_id, agenda_date, channel)``
otherwise, so a retry of the same report updates in place. A terminal
status (``executed``/``dismissed``) is never downgraded by a later
``shown`` report; ``failed`` never blocks re-delivery.
"""
def __init__(
self,
retention_days: int = 180,
max_deliveries: int = 500,
**kwargs,
):
super().__init__(**kwargs)
self.retention_days = max(int(retention_days), 1)
self.max_deliveries = max(int(max_deliveries), 1)
async def execute(self):
assert self.context is not None
started = time.monotonic()
raw_items = self.context.get("items")
items = list(raw_items) if isinstance(raw_items, (list, tuple)) else []
day = today(self, str(self.context.get("date", "") or ""))
daily = daily_dir(self)
ws = workspace_dir(self)
self.logger.info(f"[{self.name}] start day={day} items={len(items)}")
if not items:
self.context.response.success = True
self.context.response.answer = "Feedback: no delivery items reported"
self.context.response.metadata["proactive_feedback"] = {
"date": day,
"received": 0,
"added": 0,
"updated": 0,
"dropped": 0,
"trimmed": 0,
}
return self.context.response
state_file, _needs_bootstrap = load_state(ws, daily)
counts = upsert_deliveries(state_file, items, day, current_now(self))
trimmed = trim_deliveries(state_file, day, self.retention_days, self.max_deliveries)
await save_state(ws, state_file, daily)
suppressed = suppressed_delivery_ids(state_file, day, self.retention_days)
payload = {
"date": day,
"received": len(items),
"added": counts["added"],
"updated": counts["updated"],
"dropped": counts["dropped"],
"trimmed": trimmed,
"total": len(state_file.deliveries),
"suppressed_topics": sorted(suppressed),
}
self.context.response.success = True
self.context.response.answer = (
f"Feedback: added {counts['added']}, updated {counts['updated']}, "
f"dropped {counts['dropped']}, trimmed {trimmed}; "
f"{len(state_file.deliveries)} record(s) on file"
)
self.context.response.metadata["proactive_feedback"] = payload
self.context.response.metadata["duration_ms"] = int((time.monotonic() - started) * 1000)
self.logger.info(f"[{self.name}] finish {payload}")
return self.context.response

77
reme/steps/evolve/proactive/proactive.py
View file

@ -6,6 +6,11 @@ fallback) and ``horizon_days``: horizon=1 keeps the legacy single-day file
read (v1 compatible); horizon>1 reads the truth source directly, filtering by
``last_evidence_at`` recency (v5 R4 - the truth source already carries the
cross-day merge, so the reader no longer re-scans exposure products).
Agenda items the host already reported through ``proactive_feedback`` are
withheld while they sit inside ``delivery_cooldown_days``, so a consumer that
reads this job daily does not resurface the same proposal. Feedback is
optional: with an empty ``deliveries`` section nothing is withheld.
"""
import datetime as dt
@ -23,6 +28,7 @@ from .utils import (
parse_interests_topics,
sort_topics,
state_file_path,
suppressed_delivery_ids,
topic_id,
)
@ -53,11 +59,21 @@ class ProactiveStep(BaseStep):
- ``horizon_days>1`` reads the truth source and filters by evidence recency.
"""
def __init__(self, include_content: bool = True, horizon_days: int = 1, min_confidence: float = 0.4, **kwargs):
def __init__(
self,
include_content: bool = True,
horizon_days: int = 1,
min_confidence: float = 0.4,
include_delivered: bool = False,
delivery_cooldown_days: int = 3,
**kwargs,
):
super().__init__(**kwargs)
self.include_content = include_content
self.horizon_days = max(int(horizon_days), 1)
self.min_confidence = float(min_confidence)
self.include_delivered = include_delivered
self.delivery_cooldown_days = max(int(delivery_cooldown_days), 0)
async def execute(self):
assert self.context is not None
@ -70,27 +86,37 @@ class ProactiveStep(BaseStep):
min_confidence = float(raw_min_confidence)
except (TypeError, ValueError):
min_confidence = self.min_confidence
include_delivered = bool(self.context.get("include_delivered", self.include_delivered))
raw_cooldown = self.context.get("delivery_cooldown_days", self.delivery_cooldown_days)
try:
cooldown = max(int(raw_cooldown), 0)
except (TypeError, ValueError):
cooldown = self.delivery_cooldown_days
daily = self.config_value("daily_dir")
ws = workspace_dir(self)
blocked = {} if include_delivered else self._suppressed_deliveries(ws, daily, day, cooldown)
default_rel = f"{daily}/{day}/interests.yaml"
result = ProactiveResult(date=day, path=default_rel)
self.logger.info(
f"[{self.name}] start date={day} path={default_rel} include_content={include_content} "
f"horizon_days={horizon} min_confidence={min_confidence}",
f"horizon_days={horizon} min_confidence={min_confidence} "
f"include_delivered={include_delivered} delivery_cooldown_days={cooldown} blocked={len(blocked)}",
)
if horizon == 1:
outcome = self._read_single(ws, daily, day, include_content, min_confidence, result)
outcome = self._read_single(ws, daily, day, include_content, min_confidence, result, blocked)
if outcome is not None:
return outcome
# The horizon view reads open topics from the truth source and carries
# no agenda, so delivery withholding does not apply to it.
return self._read_horizon(ws, daily, day, horizon, include_content, min_confidence, result)
# ------------------------------------------------------------------
# Single-day read (legacy-compatible path)
# ------------------------------------------------------------------
def _read_single(self, ws, daily, day, include_content, min_confidence, result: ProactiveResult):
def _read_single(self, ws, daily, day, include_content, min_confidence, result: ProactiveResult, blocked=None):
rel_path = f"{daily}/{day}/interests.yaml"
abs_path = ws / daily / day / "interests.yaml"
if not abs_path.is_file():
@ -141,13 +167,14 @@ class ProactiveStep(BaseStep):
agenda_raw = data.get("agenda")
if isinstance(agenda_raw, list):
kept_ids = {str(t.get("id") or "") for t in kept}
result.agenda = [
item
for item in agenda_raw
if isinstance(item, dict) and str(item.get("topic_id") or "") in kept_ids
]
result.agenda, result.delivered_ids = self._select_agenda(agenda_raw, kept_ids, blocked)
result.summary = f"Read {len(result.topics)} proactive topic(s) from {rel_path}"
self.logger.info(f"[{self.name}] read done path={rel_path} topics={len(result.topics)}")
if result.delivered_ids:
result.summary += f"; withheld {len(result.delivered_ids)} already-delivered agenda item(s)"
self.logger.info(
f"[{self.name}] read done path={rel_path} topics={len(result.topics)} "
f"agenda={len(result.agenda)} withheld={len(result.delivered_ids)}",
)
return self._finish(True, result, include_content=include_content)
except Exception as e: # noqa: BLE001
result.error, result.summary = f"{type(e).__name__}: {e}", ""
@ -210,6 +237,35 @@ class ProactiveStep(BaseStep):
state_file, _needs_bootstrap = load_state(ws, daily)
return {str(r.get("id") or "") for r in state_file.resolved if isinstance(r, dict) and r.get("id")}
def _suppressed_deliveries(self, ws, daily: str, day: str, cooldown: int) -> dict:
"""topic_id -> blocking delivery record; empty when nothing was reported."""
state_file, _needs_bootstrap = load_state(ws, daily)
if not state_file.deliveries:
return {}
return suppressed_delivery_ids(state_file, day, cooldown)
@staticmethod
def _select_agenda(agenda_raw: list, kept_ids: set, blocked: dict | None) -> tuple[list, list]:
"""Keep agenda items whose topic survived filtering and was not delivered.
The agenda order coming out of the refresh chain *is* the priority
order (reminders pinned first), so this only ever removes items and
never re-sorts them.
"""
blocked = blocked or {}
agenda, withheld = [], []
for item in agenda_raw:
if not isinstance(item, dict):
continue
item_id = str(item.get("topic_id") or "")
if item_id not in kept_ids:
continue
if item_id in blocked:
withheld.append(item_id)
continue
agenda.append(item)
return agenda, withheld
def _finish(self, success: bool, result: ProactiveResult, *, include_content: bool):
assert self.context is not None
self.context.response.success = success
@ -222,6 +278,7 @@ class ProactiveStep(BaseStep):
"summary": result.summary,
"topics": result.topics,
**({"agenda": result.agenda} if result.agenda else {}),
**({"delivered_ids": result.delivered_ids} if result.delivered_ids else {}),
**({"content": result.content} if include_content else {}),
}
self.context.response.metadata.update(result.model_dump())

168
reme/steps/evolve/proactive/utils.py
View file

@ -17,9 +17,12 @@ from pathlib import Path
import yaml
from ....enumeration import ComponentEnum
from ....schema import ProactiveSchedule, ProactiveStateFile, ProactiveTopic
from ....schema import ProactiveDelivery, ProactiveSchedule, ProactiveStateFile, ProactiveTopic
from ....schema.proactive import (
DEFAULT_DELIVERY_CHANNEL,
DEFAULT_DELIVERY_STATUS,
DEFAULT_KIND,
TERMINAL_DELIVERY_STATUSES,
TOPIC_KINDS,
clamp_confidence,
kind_priority,
@ -114,8 +117,6 @@ def _clean_legacy_topic(raw: dict) -> dict:
# ---------------------------------------------------------------------------
# Frozen identity contract (A7 / INV-4)
# ---------------------------------------------------------------------------
def normalize_topic(title: str) -> str:
@ -134,8 +135,6 @@ def topic_id(title: str) -> str:
# ---------------------------------------------------------------------------
# Paths and material set M (F2.0)
# ---------------------------------------------------------------------------
def state_file_path(ws: Path, daily: str = "daily") -> Path:
@ -241,8 +240,6 @@ def scan_material_resource(
# ---------------------------------------------------------------------------
# Truth-source state file daily/_proactive.yaml (F1.3)
# ---------------------------------------------------------------------------
def load_state(ws: Path, daily: str = "daily") -> tuple[ProactiveStateFile, bool]:
@ -277,7 +274,7 @@ def load_state(ws: Path, daily: str = "daily") -> tuple[ProactiveStateFile, bool
logger.warning(
f"proactive state file invalid, salvaged entry by entry: {path} "
f"(dropped {dropped} unreadable entr(ies); kept open_topics={len(state.open_topics)} "
f"schedules={len(state.schedules)}; {e})",
f"schedules={len(state.schedules)} deliveries={len(state.deliveries)}; {e})",
)
return state, False
return state, needs_bootstrap
@ -286,7 +283,7 @@ def load_state(ws: Path, daily: str = "daily") -> tuple[ProactiveStateFile, bool
# Truth-source sections whose entries are validated one by one when the whole
# file fails to parse. ``resolved`` is a plain dict list, so it only needs the
# non-mapping items filtered out.
_STATE_SECTIONS = (("open_topics", ProactiveTopic), ("schedules", ProactiveSchedule))
_STATE_SECTIONS = (("open_topics", ProactiveTopic), ("schedules", ProactiveSchedule), ("deliveries", ProactiveDelivery))
def _salvage_state(data: dict) -> tuple[ProactiveStateFile, int]:
@ -431,8 +428,157 @@ def trim_state_file(state_file: ProactiveStateFile, day: str, days: int) -> None
# ---------------------------------------------------------------------------
# interests.yaml read/render (F1.2 / A2 / A4)
# ---------------------------------------------------------------------------
# Delivery feedback (host -> ReMe). Optional by design: a framework that never
# reports leaves ``deliveries`` empty and every helper below degrades to a
# no-op, so the chain behaves exactly as it did before feedback existed.
# ---------------------------------------------------------------------------
def _delivery_date(text: str) -> str:
"""``YYYY-MM-DD`` prefix of an ISO timestamp; '' when unparseable."""
raw = str(text or "").strip()
parsed = safe_date(raw[:10])
return parsed.isoformat() if parsed is not None else ""
def normalize_delivery(raw, day: str, now_dt: dt.datetime) -> ProactiveDelivery | None:
"""Coerce one host-reported item into a ``ProactiveDelivery``.
``topic_id`` is the only mandatory field; a record without one can never
match an agenda item, so it is dropped instead of being stored. Missing
``agenda_date`` defaults to the round's day and missing ``delivered_at``
to the business clock, which keeps a minimal host report
(``{"topic_id": ..., "status": ...}``) sufficient.
"""
if not isinstance(raw, dict):
return None
topic = str(raw.get("topic_id") or "").strip()
if not topic:
return None
stamp = str(raw.get("delivered_at") or "").strip()
if not _delivery_date(stamp):
stamp = now_dt.isoformat(timespec="seconds")
return ProactiveDelivery(
topic_id=topic,
agenda_date=str(raw.get("agenda_date") or day or "")[:10],
delivered_at=stamp,
channel=str(raw.get("channel") or DEFAULT_DELIVERY_CHANNEL),
consumer=str(raw.get("consumer") or ""),
status=str(raw.get("status") or DEFAULT_DELIVERY_STATUS),
outcome=str(raw.get("outcome") or ""),
external_ref=str(raw.get("external_ref") or ""),
)
def delivery_index(state_file: ProactiveStateFile) -> dict[str, list[ProactiveDelivery]]:
"""Group delivery records by topic_id, newest first.
The list is kept (not collapsed to one record) because a topic can be
surfaced through several channels; suppression only needs the most recent
relevant record, which is the head after sorting.
"""
grouped: dict[str, list[ProactiveDelivery]] = {}
for record in state_file.deliveries:
if not record.topic_id:
continue
grouped.setdefault(record.topic_id, []).append(record)
for records in grouped.values():
records.sort(key=lambda r: (str(r.delivered_at or ""), str(r.external_ref or "")), reverse=True)
return grouped
def is_delivery_suppressed(record: ProactiveDelivery | None, day: str, cooldown_days: int) -> bool:
"""Whether a delivery record still blocks re-proposing its topic.
``failed`` never blocks: a run that did not complete should be retryable
on the next round. Every other status blocks while it is younger than
``cooldown_days``, so a surfaced-but-ignored item can come back later
while an executed or dismissed one does not repeat tomorrow.
"""
if record is None:
return False
if str(record.status or "") == "failed":
return False
base = safe_date(day)
delivered = safe_date(_delivery_date(record.delivered_at))
if base is None or delivered is None:
return False
window = max(int(cooldown_days), 0)
return (base - delivered).days <= window
def suppressed_delivery_ids(
state_file: ProactiveStateFile,
day: str,
cooldown_days: int,
) -> dict[str, ProactiveDelivery]:
"""topic_id -> blocking record, for the reader and the agenda step."""
out: dict[str, ProactiveDelivery] = {}
for item_id, records in delivery_index(state_file).items():
for record in records:
if is_delivery_suppressed(record, day, cooldown_days):
out[item_id] = record
break
return out
def upsert_deliveries(
state_file: ProactiveStateFile,
items: list,
day: str,
now_dt: dt.datetime,
) -> dict[str, int]:
"""Merge host-reported deliveries into the truth source (idempotent).
Identity is ``ProactiveDelivery.identity()``: an explicit ``external_ref``
wins, otherwise ``(topic_id, agenda_date, channel)``. Status never moves
backwards out of a terminal value, so a late ``shown`` report cannot undo
an ``executed`` one.
"""
counts = {"added": 0, "updated": 0, "dropped": 0}
by_identity = {record.identity(): record for record in state_file.deliveries}
for raw in items:
record = normalize_delivery(raw, day, now_dt)
if record is None:
counts["dropped"] += 1
continue
key = record.identity()
existing = by_identity.get(key)
if existing is None:
state_file.deliveries.append(record)
by_identity[key] = record
counts["added"] += 1
continue
if existing.status in TERMINAL_DELIVERY_STATUSES and record.status not in TERMINAL_DELIVERY_STATUSES:
logger.info(f"delivery for {existing.topic_id} kept terminal status {existing.status!r}")
counts["dropped"] += 1
continue
existing.agenda_date = record.agenda_date or existing.agenda_date
existing.delivered_at = record.delivered_at
existing.channel = record.channel
existing.consumer = record.consumer or existing.consumer
existing.status = record.status
if record.outcome:
existing.outcome = record.outcome
counts["updated"] += 1
return counts
def trim_deliveries(state_file: ProactiveStateFile, day: str, retention_days: int, max_deliveries: int) -> int:
"""Prune the delivery log by age then by size; returns the dropped count."""
base = safe_date(day)
before = len(state_file.deliveries)
if base is not None:
cutoff = (base - dt.timedelta(days=max(int(retention_days), 0))).isoformat()
state_file.deliveries = [r for r in state_file.deliveries if _delivery_date(r.delivered_at) >= cutoff]
cap = max(int(max_deliveries), 0)
if len(state_file.deliveries) > cap:
state_file.deliveries.sort(key=lambda r: str(r.delivered_at or ""), reverse=True)
del state_file.deliveries[cap:]
return before - len(state_file.deliveries)
def quarantine_interests(path: Path, error: Exception) -> None:
@ -614,8 +760,6 @@ def write_interests_if_changed(ws: Path, path: Path, rendered: dict) -> bool: #
# ---------------------------------------------------------------------------
# Candidate cleaning (extract side, A3)
# ---------------------------------------------------------------------------
def clean_candidate(raw, allowed_paths: set[str], kind: str, day: str) -> dict:

620
tests/unit/test_proactive_feedback.py Normal file
View file

@ -0,0 +1,620 @@
"""Unit tests for proactive delivery feedback (host -> ReMe reporting).
Covers the ``deliveries`` truth-source section, the ``proactive_feedback``
step, the reader-side withholding in ``proactive_step`` and the write-side
suppression in ``proactive_agenda_step``. Feedback is optional by contract, so
every test that asserts new behaviour is paired with one asserting that an
empty ``deliveries`` section changes nothing.
"""
import asyncio
import datetime as dt
import yaml
from reme.components.application_context import ApplicationContext
from reme.components.runtime_context import RuntimeContext
from reme.schema import ProactiveDelivery, ProactiveStateFile
from reme.schema.proactive import MAX_DELIVERY_OUTCOME_CHARS
from reme.steps.evolve.proactive.feedback import ProactiveFeedbackStep
from reme.steps.evolve.proactive.proactive import ProactiveStep
from reme.steps.evolve.proactive.utils import (
is_delivery_suppressed,
load_state,
suppressed_delivery_ids,
topic_id,
trim_deliveries,
upsert_deliveries,
)
from .test_proactive_refresh import (
DAY,
_FileStore,
_interests,
_reply,
_run_full_chain,
_state_topic,
_topic,
_touch,
_write_state,
)
NOW = dt.datetime(2026, 8, 13, 18, 0, tzinfo=dt.timezone(dt.timedelta(hours=8)))
def _write_state_with_deliveries(ws, deliveries, open_topics=None):
"""Seed the truth source with open topics plus a delivery log."""
data = {
"version": 1,
"open_topics": open_topics or [],
"resolved": [],
"schedules": [],
"deliveries": deliveries,
}
_touch(ws / "daily" / "_proactive.yaml", yaml.safe_dump(data, allow_unicode=True, sort_keys=False))
def _write_interests_with_agenda(ws, topics, agenda, day=DAY, push=True):
"""Write a v2 exposure file carrying an agenda, as the agenda step would."""
payload = {
"version": 2,
"date": day,
"generated_at": f"{day}T18:00:00+08:00",
"push": push,
"topics": topics,
"agenda": agenda,
"suppressed": [],
}
_touch(_interests(ws, day), yaml.safe_dump(payload, allow_unicode=True, sort_keys=False))
def _agenda_item(topic_id_value, title, kind="follow_up", **extra):
item = {
"topic_id": topic_id_value,
"title": title,
"kind": kind,
"scenario_type": "resume_task",
"opener": f"对了，{title}还差一步。",
"next_action": "跑一下下一步",
"preconditions": [],
"delivery": "in_conversation",
"linked_memory": [],
"order_reason": "seeded",
}
item.update(extra)
return item
# ---------------------------------------------------------------------------
# Schema and pure helpers
# ---------------------------------------------------------------------------
def test_delivery_model_fallbacks():
"""Illegal status/channel fall back; outcome is capped; identity is stable."""
record = ProactiveDelivery(topic_id="t", status="BOGUS", channel="Inbox", outcome="x" * 900, external_ref="e1")
assert record.status == "shown"
assert record.channel == "inbox"
assert len(record.outcome) == MAX_DELIVERY_OUTCOME_CHARS
assert record.identity() == ("t", "ref:e1")
bare = ProactiveDelivery(topic_id="t", agenda_date=DAY, channel="notification")
assert bare.identity() == ("t", f"day:{DAY}|notification")
assert ProactiveDelivery(topic_id="t", channel="weird").channel == "other"
def test_upsert_deliveries_is_idempotent():
"""Re-reporting the same delivery updates in place instead of appending."""
state_file = ProactiveStateFile()
items = [
{"topic_id": "a", "status": "shown", "channel": "inbox", "external_ref": "ev1"},
{"topic_id": "b", "status": "executed", "channel": "in_conversation", "outcome": "done"},
{"topic_id": "", "status": "shown"},
"garbage",
]
assert upsert_deliveries(state_file, items, DAY, NOW) == {"added": 2, "updated": 0, "dropped": 2}
assert upsert_deliveries(state_file, items, DAY, NOW) == {"added": 0, "updated": 2, "dropped": 2}
assert len(state_file.deliveries) == 2
def test_upsert_deliveries_never_downgrades_terminal_status():
"""A late ``shown`` report cannot undo an ``executed`` one."""
state_file = ProactiveStateFile()
upsert_deliveries(state_file, [{"topic_id": "a", "status": "executed", "channel": "inbox"}], DAY, NOW)
counts = upsert_deliveries(state_file, [{"topic_id": "a", "status": "shown", "channel": "inbox"}], DAY, NOW)
assert counts == {"added": 0, "updated": 0, "dropped": 1}
assert state_file.deliveries[0].status == "executed"
def test_upsert_deliveries_defaults_missing_clock_fields():
"""A minimal report still gets an agenda_date and a delivered_at."""
state_file = ProactiveStateFile()
upsert_deliveries(state_file, [{"topic_id": "a"}], DAY, NOW)
record = state_file.deliveries[0]
assert record.agenda_date == DAY
assert record.delivered_at.startswith(DAY)
assert record.status == "shown" and record.channel == "other"
def test_is_delivery_suppressed_window():
"""``failed`` never blocks; other statuses block inside the cooldown only."""
def record(status, delivered_at):
return ProactiveDelivery(topic_id="a", status=status, delivered_at=delivered_at)
assert is_delivery_suppressed(None, DAY, 3) is False
assert is_delivery_suppressed(record("failed", f"{DAY}T18:00:00+08:00"), DAY, 3) is False
assert is_delivery_suppressed(record("shown", f"{DAY}T18:00:00+08:00"), DAY, 3) is True
assert is_delivery_suppressed(record("executed", "2026-08-10T18:00:00+08:00"), DAY, 3) is True
assert is_delivery_suppressed(record("shown", "2026-08-09T18:00:00+08:00"), DAY, 3) is False
assert is_delivery_suppressed(record("shown", "2026-08-09T18:00:00+08:00"), DAY, 0) is False
assert is_delivery_suppressed(record("shown", "not-a-date"), DAY, 3) is False
def test_trim_deliveries_by_age_then_size():
"""Old records go first, then the newest ``max_deliveries`` survive."""
state_file = ProactiveStateFile()
for index, day in enumerate(["2026-01-01", "2026-08-11", "2026-08-12", "2026-08-13"]):
state_file.deliveries.append(
ProactiveDelivery(topic_id=f"t{index}", delivered_at=f"{day}T18:00:00+08:00", external_ref=f"e{index}"),
)
assert trim_deliveries(state_file, DAY, 180, 2) == 2
assert [r.topic_id for r in state_file.deliveries] == ["t3", "t2"]
# retention 1 day keeps 08-12 and 08-13; retention 0 keeps only today.
assert trim_deliveries(state_file, DAY, 1, 10) == 0
assert trim_deliveries(state_file, DAY, 0, 10) == 1
assert [r.topic_id for r in state_file.deliveries] == ["t3"]
def test_suppressed_delivery_ids_prefers_the_blocking_record():
"""The newest blocking record wins; a failed run does not mask it."""
state_file = ProactiveStateFile()
upsert_deliveries(
state_file,
[
{"topic_id": "a", "status": "failed", "channel": "inbox", "external_ref": "e1"},
{"topic_id": "a", "status": "shown", "channel": "notification", "external_ref": "e2"},
{"topic_id": "b", "status": "failed", "channel": "inbox", "external_ref": "e3"},
],
DAY,
NOW,
)
blocked = suppressed_delivery_ids(state_file, DAY, 3)
assert set(blocked) == {"a"}
assert blocked["a"].status == "shown"
# ---------------------------------------------------------------------------
# proactive_feedback_step
# ---------------------------------------------------------------------------
def test_feedback_step_records_and_persists(tmp_path):
"""One report lands in the truth source and is summarized in the response."""
async def run():
ws = tmp_path
_write_state(ws, open_topics=[_state_topic("Topic A")])
response = await _async_feedback(
ws,
[
{
"topic_id": "a1",
"status": "executed",
"channel": "in_conversation",
"outcome": "已完成",
"external_ref": "s1",
},
{"topic_id": "a2", "status": "shown", "channel": "inbox", "external_ref": "ev1"},
],
)
assert response.success is True
assert "added 2" in response.answer
payload = response.metadata["proactive_feedback"]
assert payload["added"] == 2 and payload["updated"] == 0 and payload["total"] == 2
assert payload["suppressed_topics"] == ["a1", "a2"]
state_file, _ = load_state(ws)
assert {r.topic_id: r.status for r in state_file.deliveries} == {"a1": "executed", "a2": "shown"}
# The step must not disturb the other truth-source sections.
assert [t.title for t in state_file.open_topics] == ["Topic A"]
assert state_file.resolved == [] and state_file.schedules == []
asyncio.run(run())
async def _async_feedback(ws, items, day=DAY, **kwargs):
step = ProactiveFeedbackStep(app_context=ApplicationContext(workspace_dir=str(ws)), **kwargs)
return await step(RuntimeContext(date=day, items=items, file_store=_FileStore(ws)))
def test_feedback_step_creates_missing_truth_source(tmp_path):
"""Reporting into a fresh workspace bootstraps an otherwise empty state file."""
async def run():
response = await _async_feedback(tmp_path, [{"topic_id": "a1", "status": "shown"}])
assert response.success is True
state_file, _ = load_state(tmp_path)
assert [r.topic_id for r in state_file.deliveries] == ["a1"]
assert state_file.open_topics == []
asyncio.run(run())
def test_feedback_step_without_items_is_a_noop(tmp_path):
"""An empty report succeeds, writes nothing and says so."""
async def run():
_write_state(tmp_path, open_topics=[_state_topic("Topic A")])
before = (tmp_path / "daily" / "_proactive.yaml").read_text(encoding="utf-8")
response = await _async_feedback(tmp_path, [])
assert response.success is True
assert response.answer == "Feedback: no delivery items reported"
assert response.metadata["proactive_feedback"]["received"] == 0
assert (tmp_path / "daily" / "_proactive.yaml").read_text(encoding="utf-8") == before
asyncio.run(run())
def test_feedback_step_is_idempotent_across_runs(tmp_path):
"""Retrying the same report does not duplicate records."""
async def run():
_write_state(tmp_path)
items = [{"topic_id": "a1", "status": "shown", "channel": "inbox", "external_ref": "ev1"}]
await _async_feedback(tmp_path, items)
response = await _async_feedback(tmp_path, items)
assert "updated 1" in response.answer
state_file, _ = load_state(tmp_path)
assert len(state_file.deliveries) == 1
asyncio.run(run())
def test_feedback_step_trims_by_retention(tmp_path):
"""Records older than the retention window are pruned on write."""
async def run():
ws = tmp_path
_write_state_with_deliveries(
ws,
[{"topic_id": "old", "delivered_at": "2026-01-01T18:00:00+08:00", "status": "shown", "external_ref": "e"}],
)
response = await _async_feedback(ws, [{"topic_id": "new", "status": "shown"}], retention_days=30)
assert response.metadata["proactive_feedback"]["trimmed"] == 1
state_file, _ = load_state(ws)
assert [r.topic_id for r in state_file.deliveries] == ["new"]
asyncio.run(run())
# ---------------------------------------------------------------------------
# proactive_step (reader)
# ---------------------------------------------------------------------------
def _seed_reader(ws, fid, eid):
"""Two open topics, both on today's agenda; ``eid`` already delivered."""
follow = _state_topic("记忆检索回归", topic_id_value=fid)
extend = _state_topic("多跳 RAG 综述", topic_id_value=eid, kind="interest_extend")
_write_state_with_deliveries(
ws,
[
{
"topic_id": eid,
"agenda_date": DAY,
"delivered_at": f"{DAY}T09:00:00+08:00",
"channel": "inbox",
"status": "shown",
"external_ref": "ev1",
},
],
open_topics=[follow, extend],
)
_write_interests_with_agenda(
ws,
[follow, extend],
[_agenda_item(fid, "记忆检索回归"), _agenda_item(eid, "多跳 RAG 综述", kind="interest_extend")],
)
def test_reader_withholds_delivered_agenda_items(tmp_path):
"""A delivered agenda item disappears from the read but stays auditable."""
async def run():
ws = tmp_path
fid, eid = topic_id("记忆检索回归"), topic_id("多跳 RAG 综述")
_seed_reader(ws, fid, eid)
step = ProactiveStep(app_context=ApplicationContext(workspace_dir=str(ws)))
response = await step(RuntimeContext(date=DAY, include_content=False, file_store=_FileStore(ws)))
assert response.success is True
assert [i["topic_id"] for i in response.answer["agenda"]] == [fid]
assert response.answer["delivered_ids"] == [eid]
assert "withheld 1" in response.answer["summary"]
# Topics are material, not delivery units: they stay unfiltered.
assert {t["id"] for t in response.answer["topics"]} == {fid, eid}
assert response.metadata["delivered_ids"] == [eid]
asyncio.run(run())
def test_reader_include_delivered_returns_everything(tmp_path):
"""``include_delivered`` is the escape hatch for hosts doing their own dedup."""
async def run():
ws = tmp_path
fid, eid = topic_id("记忆检索回归"), topic_id("多跳 RAG 综述")
_seed_reader(ws, fid, eid)
step = ProactiveStep(app_context=ApplicationContext(workspace_dir=str(ws)))
response = await step(
RuntimeContext(date=DAY, include_content=False, include_delivered=True, file_store=_FileStore(ws)),
)
assert [i["topic_id"] for i in response.answer["agenda"]] == [fid, eid]
assert "delivered_ids" not in response.answer
asyncio.run(run())
def test_reader_cooldown_expiry_lets_an_item_back(tmp_path):
"""A merely-shown item returns once the cooldown has passed."""
async def run():
ws = tmp_path
fid, eid = topic_id("记忆检索回归"), topic_id("多跳 RAG 综述")
follow = _state_topic("记忆检索回归", topic_id_value=fid)
extend = _state_topic("多跳 RAG 综述", topic_id_value=eid, kind="interest_extend")
# The single-day exposure is read at DAY, so the delivery is backdated
# instead: shown a week ago, default cooldown is three days.
_write_state_with_deliveries(
ws,
[
{
"topic_id": eid,
"agenda_date": "2026-08-06",
"delivered_at": "2026-08-06T09:00:00+08:00",
"channel": "inbox",
"status": "shown",
},
],
open_topics=[follow, extend],
)
_write_interests_with_agenda(
ws,
[follow, extend],
[_agenda_item(fid, "记忆检索回归"), _agenda_item(eid, "多跳 RAG 综述", kind="interest_extend")],
)
step = ProactiveStep(app_context=ApplicationContext(workspace_dir=str(ws)))
response = await step(RuntimeContext(date=DAY, include_content=False, file_store=_FileStore(ws)))
assert [i["topic_id"] for i in response.answer["agenda"]] == [fid, eid]
assert "delivered_ids" not in response.answer
asyncio.run(run())
def test_reader_without_feedback_is_unchanged(tmp_path):
"""No ``deliveries`` section means no withholding (backwards compatible)."""
async def run():
ws = tmp_path
fid, eid = topic_id("记忆检索回归"), topic_id("多跳 RAG 综述")
follow = _state_topic("记忆检索回归", topic_id_value=fid)
extend = _state_topic("多跳 RAG 综述", topic_id_value=eid, kind="interest_extend")
_write_state(ws, open_topics=[follow, extend])
_write_interests_with_agenda(
ws,
[follow, extend],
[_agenda_item(fid, "记忆检索回归"), _agenda_item(eid, "多跳 RAG 综述", kind="interest_extend")],
)
step = ProactiveStep(app_context=ApplicationContext(workspace_dir=str(ws)))
response = await step(RuntimeContext(date=DAY, include_content=False, file_store=_FileStore(ws)))
assert [i["topic_id"] for i in response.answer["agenda"]] == [fid, eid]
assert "delivered_ids" not in response.answer
assert "withheld" not in response.answer["summary"]
asyncio.run(run())
def test_reader_reminder_still_outranks_and_survives(tmp_path):
"""Withholding a discovery item never reorders or drops a pinned reminder."""
async def run():
ws = tmp_path
rid, eid = topic_id("牙医预约"), topic_id("多跳 RAG 综述")
reminder = _state_topic("牙医预约", topic_id_value=rid, kind="reminder", confidence=0.9)
extend = _state_topic("多跳 RAG 综述", topic_id_value=eid, kind="interest_extend")
_write_state_with_deliveries(
ws,
[
{
"topic_id": eid,
"agenda_date": DAY,
"delivered_at": f"{DAY}T09:00:00+08:00",
"channel": "inbox",
"status": "shown",
},
],
open_topics=[reminder, extend],
)
_write_interests_with_agenda(
ws,
[reminder, extend],
[
_agenda_item(rid, "牙医预约", kind="reminder", scenario_type="prepare_upcoming"),
_agenda_item(eid, "多跳 RAG 综述", kind="interest_extend"),
],
)
step = ProactiveStep(app_context=ApplicationContext(workspace_dir=str(ws)))
response = await step(RuntimeContext(date=DAY, include_content=False, file_store=_FileStore(ws)))
assert [i["topic_id"] for i in response.answer["agenda"]] == [rid]
assert response.answer["agenda"][0]["kind"] == "reminder"
asyncio.run(run())
# ---------------------------------------------------------------------------
# proactive_agenda_step (writer)
# ---------------------------------------------------------------------------
def test_agenda_step_suppresses_delivered_candidate(tmp_path):
"""A reported candidate leaves the agenda with an auditable reason."""
async def run():
ws = tmp_path
_touch(ws / "daily" / DAY / "session.md", "we discussed the eval plan")
_write_state_with_deliveries(
ws,
[
{
"topic_id": topic_id("任务一"),
"agenda_date": DAY,
"delivered_at": f"{DAY}T09:00:00+08:00",
"channel": "in_conversation",
"status": "executed",
"external_ref": "s1",
},
],
)
extract_reply = _reply(
follow_ups=[
_topic("任务一", confidence=0.9, paths=[f"daily/{DAY}/session.md"]),
_topic("任务二", confidence=0.8, paths=[f"daily/{DAY}/session.md"]),
],
)
_, _, _, context, resps = await _run_full_chain(ws, [extract_reply])
assert all(r.success for r in resps)
state = context.get("proactive")
assert state["suppressed_delivered"] == 1
data = yaml.safe_load(_interests(ws).read_text(encoding="utf-8"))
assert [i["title"] for i in data["agenda"]] == ["任务二"]
suppressed = {s["topic_id"]: s["reason"] for s in data["suppressed"]}
reason = suppressed[topic_id("任务一")]
assert reason.startswith("already delivered on")
assert "in_conversation" in reason and "executed" in reason
asyncio.run(run())
def test_agenda_step_without_feedback_is_unchanged(tmp_path):
"""No delivery log means the deterministic fallback agenda stays intact."""
async def run():
ws = tmp_path
_touch(ws / "daily" / DAY / "session.md", "material")
extract_reply = _reply(
follow_ups=[
_topic("任务一", confidence=0.9, paths=[f"daily/{DAY}/session.md"]),
_topic("任务二", confidence=0.7, paths=[f"daily/{DAY}/session.md"]),
],
)
_, _, _, context, resps = await _run_full_chain(ws, [extract_reply])
assert all(r.success for r in resps)
assert context.get("proactive")["suppressed_delivered"] == 0
data = yaml.safe_load(_interests(ws).read_text(encoding="utf-8"))
assert [i["title"] for i in data["agenda"]] == ["任务一", "任务二"]
assert data["suppressed"] == []
asyncio.run(run())
def test_agenda_step_cooldown_zero_only_blocks_same_day(tmp_path):
"""``redelivery_cooldown_days=0`` still blocks a delivery reported today."""
async def run():
ws = tmp_path
_touch(ws / "daily" / DAY / "session.md", "material")
_write_state_with_deliveries(
ws,
[
{
"topic_id": topic_id("任务一"),
"agenda_date": DAY,
"delivered_at": f"{DAY}T09:00:00+08:00",
"channel": "inbox",
"status": "shown",
},
],
)
extract_reply = _reply(follow_ups=[_topic("任务一", confidence=0.9, paths=[f"daily/{DAY}/session.md"])])
_, _, _, context, resps = await _run_full_chain(
ws,
[extract_reply],
agenda_kwargs={"redelivery_cooldown_days": 0},
)
assert all(r.success for r in resps)
assert context.get("proactive")["suppressed_delivered"] == 1
data = yaml.safe_load(_interests(ws).read_text(encoding="utf-8"))
assert data["agenda"] == []
assert data["push"] is False
asyncio.run(run())
# ---------------------------------------------------------------------------
# Host interop: the report shape a consumer actually sends
# ---------------------------------------------------------------------------
def test_consumer_report_shape_round_trips(tmp_path):
"""A host report in the wire shape the consumer sends is stored verbatim and withholds.
Locks the interop contract on three points that are easy to break silently:
every field of the report survives (an unknown ``consumer`` label must not be
dropped, a known ``channel`` must not be coerced to ``other``), a UTC
``delivered_at`` still suppresses on a later local business day, and the
reader publishes ``delivered_ids`` in metadata for the host to count.
"""
async def run():
ws = tmp_path
fid, eid = topic_id("记忆检索回归"), topic_id("多跳 RAG 综述")
follow = _state_topic("记忆检索回归", topic_id_value=fid)
extend = _state_topic("多跳 RAG 综述", topic_id_value=eid, kind="interest_extend")
_write_state(ws, open_topics=[follow, extend])
_write_interests_with_agenda(
ws,
[follow, extend],
[_agenda_item(fid, "记忆检索回归"), _agenda_item(eid, "多跳 RAG 综述", kind="interest_extend")],
)
# 23:30Z on the previous UTC day is 07:30 on DAY in +08:00: the consumer
# stamps UTC, the business clock is local, so the date prefix lags.
report = {
"topic_id": eid,
"agenda_date": DAY,
"delivered_at": "2026-08-12T23:30:00.123456+00:00",
"channel": "in_conversation",
"consumer": "qwenpaw",
"status": "shown",
"outcome": "",
"external_ref": "session-7f3a",
}
response = await _async_feedback(ws, [report])
assert response.success is True and "added 1" in response.answer
state_file, _ = load_state(ws)
assert len(state_file.deliveries) == 1
stored = state_file.deliveries[0]
assert stored.consumer == "qwenpaw" # a free-form host label is kept as sent
assert stored.channel == "in_conversation" # in the vocabulary, so not coerced
assert stored.external_ref == "session-7f3a"
assert stored.agenda_date == DAY
assert stored.delivered_at.startswith("2026-08-12T23:30:00")
assert is_delivery_suppressed(stored, DAY, 3) is True
# Re-reporting the same delivery updates in place instead of piling up.
again = await _async_feedback(ws, [dict(report, status="executed")])
assert again.success is True
state_file, _ = load_state(ws)
assert len(state_file.deliveries) == 1
assert state_file.deliveries[0].status == "executed"
reader = ProactiveStep(app_context=ApplicationContext(workspace_dir=str(ws)))
read = await reader(RuntimeContext(date=DAY, include_content=False, file_store=_FileStore(ws)))
assert read.success is True
assert [item["topic_id"] for item in read.answer["agenda"]] == [fid]
assert read.metadata["delivered_ids"] == [eid]
asyncio.run(run())

8
tests/unit/test_proactive_refresh.py
View file

@ -2933,6 +2933,11 @@ def test_state_file_bare_yaml_dates_load(tmp_path):
" - at: 2026-07-01\n"
" action: add\n"
" to: 2026-08-05T09:00:00\n"
"deliveries:\n"
" - topic_id: t1\n"
" agenda_date: 2026-08-12\n"
" delivered_at: 2026-08-12T09:00:00\n"
" status: shown\n"
)
_touch(tmp_path / "daily" / "_proactive.yaml", raw)
@ -2946,6 +2951,9 @@ def test_state_file_bare_yaml_dates_load(tmp_path):
assert (schedule.first_seen, schedule.last_seen_at) == ("2026-07-01", "2026-07-01")
assert schedule.history[0]["at"] == "2026-07-01"
assert schedule.history[0]["to"] == "2026-08-05T09:00:00"
delivery = state_file.deliveries[0]
assert delivery.agenda_date == "2026-08-12"
assert delivery.delivered_at == "2026-08-12T09:00:00"
def test_state_file_salvages_unreadable_entries(tmp_path):