From a0f66c17790bd95a71b7e9ce4e39e48198440f9d Mon Sep 17 00:00:00 2001 From: imrewce Date: Mon, 12 Oct 2026 15:36:56 +0800 Subject: [PATCH] feat(proactive): accept host delivery receipts 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. --- reme/config/default.yaml | 33 ++ reme/schema/__init__.py | 2 + reme/schema/proactive.py | 80 +++ reme/steps/evolve/proactive/__init__.py | 2 + reme/steps/evolve/proactive/agenda.py | 53 +- reme/steps/evolve/proactive/feedback.py | 106 ++++ reme/steps/evolve/proactive/proactive.py | 77 ++- reme/steps/evolve/proactive/utils.py | 168 +++++- tests/unit/test_proactive_feedback.py | 620 +++++++++++++++++++++++ tests/unit/test_proactive_refresh.py | 8 + 10 files changed, 1124 insertions(+), 25 deletions(-) create mode 100644 reme/steps/evolve/proactive/feedback.py create mode 100644 tests/unit/test_proactive_feedback.py diff --git a/reme/config/default.yaml b/reme/config/default.yaml index a29f4901..427eceb1 100644 --- a/reme/config/default.yaml +++ b/reme/config/default.yaml @@ -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 diff --git a/reme/schema/__init__.py b/reme/schema/__init__.py index 5dd46c5a..b98f3909 100644 --- a/reme/schema/__init__.py +++ b/reme/schema/__init__.py @@ -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", diff --git a/reme/schema/proactive.py b/reme/schema/proactive.py index 06426304..d4beb1e6 100644 --- a/reme/schema/proactive.py +++ b/reme/schema/proactive.py @@ -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) diff --git a/reme/steps/evolve/proactive/__init__.py b/reme/steps/evolve/proactive/__init__.py index 508583ae..e58a77c2 100644 --- a/reme/steps/evolve/proactive/__init__.py +++ b/reme/steps/evolve/proactive/__init__.py @@ -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", diff --git a/reme/steps/evolve/proactive/agenda.py b/reme/steps/evolve/proactive/agenda.py index 5f57d0b4..a4edda68 100644 --- a/reme/steps/evolve/proactive/agenda.py +++ b/reme/steps/evolve/proactive/agenda.py @@ -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")) diff --git a/reme/steps/evolve/proactive/feedback.py b/reme/steps/evolve/proactive/feedback.py new file mode 100644 index 00000000..e5a06e0a --- /dev/null +++ b/reme/steps/evolve/proactive/feedback.py @@ -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 diff --git a/reme/steps/evolve/proactive/proactive.py b/reme/steps/evolve/proactive/proactive.py index a5a3ec71..bcbeaff7 100644 --- a/reme/steps/evolve/proactive/proactive.py +++ b/reme/steps/evolve/proactive/proactive.py @@ -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()) diff --git a/reme/steps/evolve/proactive/utils.py b/reme/steps/evolve/proactive/utils.py index 9a771b7f..0eb55882 100644 --- a/reme/steps/evolve/proactive/utils.py +++ b/reme/steps/evolve/proactive/utils.py @@ -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: diff --git a/tests/unit/test_proactive_feedback.py b/tests/unit/test_proactive_feedback.py new file mode 100644 index 00000000..a79eb2a7 --- /dev/null +++ b/tests/unit/test_proactive_feedback.py @@ -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()) diff --git a/tests/unit/test_proactive_refresh.py b/tests/unit/test_proactive_refresh.py index 31197c67..0ecb392d 100644 --- a/tests/unit/test_proactive_refresh.py +++ b/tests/unit/test_proactive_refresh.py @@ -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):