Session-purity invariant + card_refresh dream type (DEV-2000) (#883)
* fix: enforce explicit-document session purity in dedup/merge paths
Audit for DEV-2000 (Scopes RFC prerequisite): explicit-level documents must
stay session-pure so scope memory can be built by copying explicit documents
between collections. Two classes of violation were possible:
- Exact-content and semantic dedup in crud/document.py matched candidates
with no level or session scoping, so an explicit document could be
reinforced by — or soft-deleted in favor of — a same-content document from
a different session or a different level (silently merging cross-session
derivations into one row).
- The generic create_observations tool handler accepted level='explicit'
from agents with no message context (dreamer/dialectic), which would mint
session-less explicit documents.
Enforcement (refuse, never rewrite):
- create_documents refuses explicit documents with a null session_name
- exact dedup keys on (content, level, session-for-explicit); derived levels
keep cross-session consolidation
- is_rejected_duplicate scopes candidate search to the same level, and the
same session for explicit documents
- the create_observations tool rejects explicit-level input outside message
ingestion (deriver) context
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
* feat: add card_refresh dream type for event-driven peer-card updates
Adds a lightweight dream variant (DEV-2000, Scopes RFC prerequisite) that
runs ONLY the peer-card update — for event-driven refreshes such as scope
membership changes and cold starts:
- DreamType.CARD_REFRESH alongside OMNI; dispatched by process_dream to a
new run_card_refresh_dream orchestration
- CardRefreshSpecialist: restricted to get_recent_observations,
search_memory, and update_peer_card (no observation-mutating tools), with
a low tool-iteration cap of min(6, DREAM.MAX_TOOL_ITERATIONS)
- rebuild=True mode carried in the dream payload: the existing card is NOT
injected into the prompt and the specialist rebuilds it solely from
observations present in the collection (for use after removals)
- enqueue-able via the manual enqueue_dream path (bypasses volume gates);
the work-unit key already embeds the dream type so a card refresh never
collides with a pending omni dream. POST /v3/workspaces/{id}/schedule_dream
accepts dream_type=card_refresh plus the rebuild flag
- card refreshes never advance the omni dream guard pair
(last_dream_at / last_dream_document_count)
- shared PEER CARD prompt section extracted (verbatim) from
DeductionSpecialist for reuse; CallPurpose gains dream.card_refresh
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
* chore: fix tests
---------
Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
Vineeth Voruganti
GitHub
parent
4f9a41360a
commit
a15c782985
|
|
@ -443,6 +443,30 @@ def _normalize_content(content: str) -> str:
|
|||
return content.strip().lower()
|
||||
|
||||
|
||||
def _dedup_key(
|
||||
content: str, level: str, session_name: str | None
|
||||
) -> tuple[str, str, str | None]:
|
||||
"""Build the exact-match dedup key for a document.
|
||||
|
||||
Dedup never crosses levels: a same-content document at a different level is
|
||||
a different kind of record (an explicit fact is not interchangeable with a
|
||||
deductive conclusion that happens to share its text).
|
||||
|
||||
For **explicit** documents dedup additionally never crosses sessions.
|
||||
Explicit documents are session-pure records of what was derived from that
|
||||
session's messages — the Scopes copy-by-session model depends on this — so
|
||||
a repeat of the same fact in a different session must produce a new
|
||||
document in that session rather than reinforce another session's row.
|
||||
Derived levels (deductive/inductive/contradiction) are consolidations and
|
||||
may still dedup across sessions.
|
||||
"""
|
||||
return (
|
||||
_normalize_content(content),
|
||||
level,
|
||||
session_name if level == "explicit" else None,
|
||||
)
|
||||
|
||||
|
||||
@dataclass
|
||||
class CreateDocumentsResult:
|
||||
created_documents: list[schemas.DocumentCreate] = field(default_factory=list)
|
||||
|
|
@ -488,9 +512,10 @@ async def create_documents(
|
|||
# exact-content dedup (independent of `deduplicate`): pre-fetch
|
||||
# existing live documents whose normalized content matches anything in this
|
||||
# batch, scoped to (workspace, observer, observed). The SQL normalization must
|
||||
# mirror _normalize_content.
|
||||
# mirror _normalize_content. Matching is further scoped per-document by
|
||||
# level (always) and session (for explicit documents) via _dedup_key.
|
||||
batch_normalized: set[str] = {_normalize_content(d.content) for d in documents}
|
||||
existing_by_normalized: dict[str, models.Document] = {}
|
||||
existing_by_key: dict[tuple[str, str, str | None], models.Document] = {}
|
||||
if batch_normalized:
|
||||
# The `normalized_content_sql.in_(...)` filter below narrows to the
|
||||
# (workspace, observer, observed) partition via the single-column indexes,
|
||||
|
|
@ -518,15 +543,20 @@ async def create_documents(
|
|||
)
|
||||
)
|
||||
for existing_doc in existing_result.scalars():
|
||||
# If multiple historical rows share normalized content, reinforcing
|
||||
# If multiple historical rows share a dedup key, reinforcing
|
||||
# one is sufficient; keep the first.
|
||||
existing_by_normalized.setdefault(
|
||||
_normalize_content(existing_doc.content), existing_doc
|
||||
existing_by_key.setdefault(
|
||||
_dedup_key(
|
||||
existing_doc.content,
|
||||
existing_doc.level,
|
||||
existing_doc.session_name,
|
||||
),
|
||||
existing_doc,
|
||||
)
|
||||
|
||||
# Tracks normalized content already accepted from this batch so exact
|
||||
# Tracks dedup keys already accepted from this batch so exact
|
||||
# duplicates within a single inference call collapse to one document.
|
||||
seen_in_batch: set[str] = set()
|
||||
seen_in_batch: set[tuple[str, str, str | None]] = set()
|
||||
|
||||
exact_dup_existing_count = 0
|
||||
exact_dup_in_batch_count = 0
|
||||
|
|
@ -534,18 +564,33 @@ async def create_documents(
|
|||
semantic_dup_replaced_count = 0
|
||||
for doc in documents:
|
||||
try:
|
||||
normalized_content = _normalize_content(doc.content)
|
||||
# Session-purity invariant: an explicit document must always carry
|
||||
# the session it was derived from. Refuse to write session-less
|
||||
# explicit documents rather than silently minting global explicit
|
||||
# memory (the Scopes copy-by-session model depends on explicit
|
||||
# documents staying session-pure).
|
||||
if doc.level == "explicit" and doc.session_name is None:
|
||||
logger.error(
|
||||
"Refusing to create explicit document without session_name in %s/%s/%s (session-purity invariant): %r",
|
||||
workspace_name,
|
||||
observer,
|
||||
observed,
|
||||
doc.content[:80],
|
||||
)
|
||||
continue
|
||||
|
||||
dedup_key = _dedup_key(doc.content, doc.level, doc.session_name)
|
||||
|
||||
# Exact-match dedup, always on:
|
||||
# 1) collapse exact duplicates within this batch (drop silently).
|
||||
if normalized_content in seen_in_batch:
|
||||
if dedup_key in seen_in_batch:
|
||||
exact_dup_in_batch_count += 1
|
||||
continue
|
||||
seen_in_batch.add(normalized_content)
|
||||
seen_in_batch.add(dedup_key)
|
||||
|
||||
# 2) drop exact duplicates of an existing live document, recording
|
||||
# the re-derivation as reinforcement on the existing row.
|
||||
existing_match = existing_by_normalized.get(normalized_content)
|
||||
existing_match = existing_by_key.get(dedup_key)
|
||||
if existing_match is not None:
|
||||
# Reinforce the existing row. greatest(...) keeps the bump atomic
|
||||
# server-side (concurrent workers can't lose an increment) while
|
||||
|
|
@ -1114,7 +1159,20 @@ async def is_rejected_duplicate(
|
|||
If the document is a duplicate AND the existing document is superior,
|
||||
increments the existing document's ``times_derived`` to record the
|
||||
reinforcement, then returns True.
|
||||
|
||||
Merges are scoped so they never cross document levels, and never cross
|
||||
sessions for explicit-level documents (session-purity invariant: an
|
||||
explicit document records what was derived from exactly one session, so
|
||||
a near-duplicate from another session must not reinforce or replace it).
|
||||
"""
|
||||
filters: dict[str, Any] = {"level": doc.level}
|
||||
if doc.level == "explicit":
|
||||
if doc.session_name is None:
|
||||
# create_documents refuses session-less explicit documents; if one
|
||||
# reaches here anyway it has no valid merge partner.
|
||||
return SemanticRejectionResult.NOT_DUPLICATE
|
||||
filters["session_name"] = doc.session_name
|
||||
|
||||
# Step 1: Find potential duplicates using cosine similarity
|
||||
similar_docs = await query_documents(
|
||||
db=db,
|
||||
|
|
@ -1122,6 +1180,7 @@ async def is_rejected_duplicate(
|
|||
query=doc.content,
|
||||
observer=observer,
|
||||
observed=observed,
|
||||
filters=filters,
|
||||
max_distance=0.05,
|
||||
top_k=1,
|
||||
embedding=doc.embedding,
|
||||
|
|
|
|||
|
|
@ -403,6 +403,7 @@ def create_dream_record(
|
|||
delay_reason: str | None = None,
|
||||
documents_since_last_dream_at_schedule: int | None = None,
|
||||
document_threshold: int | None = None,
|
||||
rebuild: bool = False,
|
||||
) -> dict[str, Any]:
|
||||
"""
|
||||
Create a queue record for a dream task.
|
||||
|
|
@ -417,6 +418,7 @@ def create_dream_record(
|
|||
delay_reason: what governed when it fires
|
||||
documents_since_last_dream_at_schedule: count snapshot at schedule time
|
||||
document_threshold: DOCUMENT_THRESHOLD snapshot at schedule time
|
||||
rebuild: card_refresh only — rebuild the card without the prior card
|
||||
|
||||
Returns:
|
||||
Queue record dictionary with workspace_name and other fields
|
||||
|
|
@ -430,6 +432,7 @@ def create_dream_record(
|
|||
delay_reason=delay_reason,
|
||||
documents_since_last_dream_at_schedule=documents_since_last_dream_at_schedule,
|
||||
document_threshold=document_threshold,
|
||||
rebuild=rebuild,
|
||||
)
|
||||
|
||||
return {
|
||||
|
|
@ -452,6 +455,7 @@ async def enqueue_dream(
|
|||
delay_reason: str | None = None,
|
||||
documents_since_last_dream_at_schedule: int | None = None,
|
||||
document_threshold: int | None = None,
|
||||
rebuild: bool = False,
|
||||
) -> None:
|
||||
"""
|
||||
Enqueue a dream task for immediate processing by the deriver.
|
||||
|
|
@ -461,6 +465,8 @@ async def enqueue_dream(
|
|||
|
||||
Deduplication: If a dream with the same work_unit_key is already in-progress
|
||||
(has an ActiveQueueSession) or pending in the queue, the enqueue is skipped.
|
||||
The work unit key includes the dream type, so e.g. a card_refresh dream
|
||||
never collides with a pending omni dream for the same collection.
|
||||
|
||||
Args:
|
||||
workspace_name: Name of the workspace
|
||||
|
|
@ -468,6 +474,7 @@ async def enqueue_dream(
|
|||
observed: Name of the observed peer
|
||||
dream_type: Type of dream to execute
|
||||
session_name: Name of the session to scope the dream to if specified
|
||||
rebuild: card_refresh only — rebuild the card without the prior card
|
||||
"""
|
||||
async with tracked_db("dream_enqueue") as db_session:
|
||||
try:
|
||||
|
|
@ -481,6 +488,7 @@ async def enqueue_dream(
|
|||
delay_reason=delay_reason,
|
||||
documents_since_last_dream_at_schedule=documents_since_last_dream_at_schedule,
|
||||
document_threshold=document_threshold,
|
||||
rebuild=rebuild,
|
||||
)
|
||||
|
||||
work_unit_key = dream_record["work_unit_key"]
|
||||
|
|
|
|||
|
|
@ -26,7 +26,11 @@ from sqlalchemy import func, select
|
|||
from src import crud, models
|
||||
from src.config import settings
|
||||
from src.dependencies import tracked_db
|
||||
from src.dreamer.specialists import SPECIALISTS, SpecialistResult
|
||||
from src.dreamer.specialists import (
|
||||
SPECIALISTS,
|
||||
CardRefreshSpecialist,
|
||||
SpecialistResult,
|
||||
)
|
||||
from src.dreamer.surprisal import SurprisalScore # type: ignore
|
||||
from src.exceptions import SurprisalError
|
||||
from src.schemas import DreamType
|
||||
|
|
@ -307,6 +311,151 @@ async def run_dream(
|
|||
)
|
||||
|
||||
|
||||
async def run_card_refresh_dream(
|
||||
workspace_name: str,
|
||||
observer: str,
|
||||
observed: str,
|
||||
session_name: str | None = None,
|
||||
*,
|
||||
rebuild: bool = False,
|
||||
dream_type: str | None = None,
|
||||
trigger_reason: str | None = None,
|
||||
delay_reason: str | None = None,
|
||||
) -> DreamResult | None:
|
||||
"""
|
||||
Run a lightweight card-only refresh dream.
|
||||
|
||||
Runs a single CardRefreshSpecialist restricted to peer-card tools
|
||||
(get_recent_observations, search_memory, update_peer_card) with a low
|
||||
tool-iteration cap. It never creates or deletes observations.
|
||||
|
||||
Args:
|
||||
workspace_name: Workspace identifier
|
||||
observer: Observer peer name
|
||||
observed: Observed peer name
|
||||
session_name: Session identifier if specified
|
||||
rebuild: When True the existing card is NOT injected into the prompt
|
||||
and the specialist rebuilds it solely from observations present in
|
||||
the collection (used after removals).
|
||||
"""
|
||||
if not settings.DREAM.ENABLED:
|
||||
return None
|
||||
|
||||
run_id = generate_nanoid()
|
||||
task_name = f"dream_orchestrator_{run_id}"
|
||||
start_time = time.perf_counter()
|
||||
|
||||
logger.info(
|
||||
f"[{run_id}] Starting card-refresh dream for {workspace_name}/{observer}/{observed} (rebuild={rebuild})"
|
||||
)
|
||||
|
||||
# Short-lived DB session for config resolution
|
||||
async with tracked_db("dream.config") as db:
|
||||
if session_name is not None:
|
||||
session = await crud.get_session(
|
||||
db, workspace_name=workspace_name, session_name=session_name
|
||||
)
|
||||
else:
|
||||
session = None
|
||||
|
||||
workspace = await crud.get_workspace(db, workspace_name=workspace_name)
|
||||
configuration = get_configuration(None, session, workspace)
|
||||
if not configuration.dream.enabled:
|
||||
logger.info(
|
||||
f"[{run_id}] Dreams disabled for {workspace_name}/{session_name}, skipping card refresh"
|
||||
)
|
||||
return None
|
||||
if not configuration.peer_card.create:
|
||||
logger.info(
|
||||
f"[{run_id}] Peer card creation disabled for {workspace_name}, skipping card refresh"
|
||||
)
|
||||
return None
|
||||
|
||||
specialist_success = False
|
||||
specialist_result: SpecialistResult | None = None
|
||||
duration_ms = 0.0
|
||||
try:
|
||||
specialist = CardRefreshSpecialist(rebuild=rebuild)
|
||||
try:
|
||||
specialist_result = await specialist.run(
|
||||
workspace_name=workspace_name,
|
||||
observer=observer,
|
||||
observed=observed,
|
||||
session_name=session_name,
|
||||
configuration=configuration,
|
||||
parent_run_id=run_id,
|
||||
)
|
||||
logger.info(
|
||||
f"[{run_id}] Card refresh completed: {specialist_result.content[:200]}..."
|
||||
)
|
||||
accumulate_metric(
|
||||
task_name, "card_refresh_result", specialist_result.content, "blob"
|
||||
)
|
||||
specialist_success = specialist_result.success
|
||||
except Exception as e:
|
||||
# Exception (not BaseException) — CancelledError must propagate so
|
||||
# the worker can shut down; the finally still emits the run event.
|
||||
logger.error(
|
||||
f"[{run_id}] Card refresh specialist failed: {e}", exc_info=True
|
||||
)
|
||||
accumulate_metric(task_name, "card_refresh_error", str(e), "blob")
|
||||
|
||||
duration_ms = (time.perf_counter() - start_time) * 1000
|
||||
accumulate_metric(task_name, "total_duration", duration_ms, "ms")
|
||||
logger.info(f"[{run_id}] Card-refresh dream completed in {duration_ms:.0f}ms")
|
||||
log_performance_metrics("dream_orchestrator", run_id)
|
||||
finally:
|
||||
# Emit DreamRunEvent unconditionally so analytics see a parent for the
|
||||
# specialist event, mirroring run_dream. Card refresh is a
|
||||
# deduction-family run, so its outcome rides on deduction_success.
|
||||
if duration_ms == 0.0:
|
||||
duration_ms = (time.perf_counter() - start_time) * 1000
|
||||
try:
|
||||
emit(
|
||||
DreamRunEvent(
|
||||
run_id=run_id,
|
||||
workspace_name=workspace_name,
|
||||
session_name=session_name,
|
||||
observer=observer,
|
||||
observed=observed,
|
||||
specialists_run=["card_refresh"],
|
||||
deduction_success=specialist_success,
|
||||
induction_success=False,
|
||||
surprisal_enabled=False,
|
||||
surprisal_conclusion_count=0,
|
||||
total_iterations=(
|
||||
specialist_result.iterations if specialist_result else 0
|
||||
),
|
||||
total_input_tokens=(
|
||||
specialist_result.input_tokens if specialist_result else 0
|
||||
),
|
||||
total_output_tokens=(
|
||||
specialist_result.output_tokens if specialist_result else 0
|
||||
),
|
||||
total_duration_ms=duration_ms,
|
||||
dream_type=dream_type,
|
||||
enabled_types_count=len(settings.DREAM.ENABLED_TYPES),
|
||||
trigger_reason=trigger_reason,
|
||||
delay_reason=delay_reason,
|
||||
)
|
||||
)
|
||||
except Exception: # pragma: no cover - telemetry must not raise
|
||||
logger.debug("Failed to emit DreamRunEvent", exc_info=True)
|
||||
|
||||
return DreamResult(
|
||||
run_id=run_id,
|
||||
specialists_run=["card_refresh"],
|
||||
deduction_success=specialist_success,
|
||||
induction_success=False,
|
||||
surprisal_enabled=False,
|
||||
surprisal_conclusion_count=0,
|
||||
total_iterations=specialist_result.iterations if specialist_result else 0,
|
||||
total_duration_ms=duration_ms,
|
||||
input_tokens=specialist_result.input_tokens if specialist_result else 0,
|
||||
output_tokens=specialist_result.output_tokens if specialist_result else 0,
|
||||
)
|
||||
|
||||
|
||||
def _create_queries_from_surprisal(
|
||||
high_surprisal_obs: list[SurprisalScore],
|
||||
) -> list[str]:
|
||||
|
|
@ -401,6 +550,28 @@ DREAM: {payload.dream_type} documents for {workspace_name}/{payload.observer}/{p
|
|||
update_data={"dream": dream_meta},
|
||||
)
|
||||
|
||||
case DreamType.CARD_REFRESH:
|
||||
# Card-only refresh: never touches observations and never
|
||||
# advances the omni dream guard pair (last_dream_at /
|
||||
# last_dream_document_count) — a card refresh must not delay
|
||||
# or satisfy consolidation scheduling.
|
||||
result = await run_card_refresh_dream(
|
||||
workspace_name=workspace_name,
|
||||
observer=payload.observer,
|
||||
observed=payload.observed,
|
||||
session_name=payload.session_name,
|
||||
rebuild=payload.rebuild,
|
||||
dream_type=payload.dream_type.value,
|
||||
trigger_reason=payload.trigger_reason,
|
||||
delay_reason=payload.delay_reason,
|
||||
)
|
||||
if result is not None:
|
||||
logger.info(
|
||||
f"Card-refresh dream completed: run_id={result.run_id}, "
|
||||
+ f"iterations={result.total_iterations}, "
|
||||
+ f"duration={result.total_duration_ms:.0f}ms"
|
||||
)
|
||||
|
||||
except Exception as e:
|
||||
logger.error(
|
||||
f"Error processing dream task {payload.dream_type} for {payload.observer}/{payload.observed}: {str(e)}",
|
||||
|
|
|
|||
|
|
@ -32,6 +32,7 @@ from src.telemetry.events import DreamSpecialistEvent, emit
|
|||
from src.telemetry.logging import accumulate_metric, log_performance_metrics
|
||||
from src.telemetry.prometheus.metrics import TokenTypes
|
||||
from src.utils.agent_tools import (
|
||||
CARD_REFRESH_SPECIALIST_TOOLS,
|
||||
DEDUCTION_SPECIALIST_TOOLS,
|
||||
INDUCTION_SPECIALIST_TOOLS,
|
||||
create_tool_executor,
|
||||
|
|
@ -70,6 +71,64 @@ class SpecialistResult:
|
|||
# Tool names to exclude when peer card creation is disabled
|
||||
PEER_CARD_TOOL_NAMES = {"update_peer_card"}
|
||||
|
||||
# Shared PEER CARD system-prompt section (identity-store taxonomy + rules).
|
||||
# Used verbatim by DeductionSpecialist and CardRefreshSpecialist.
|
||||
PEER_CARD_SYSTEM_SECTION = """
|
||||
|
||||
## PEER CARD (REQUIRED)
|
||||
|
||||
The peer card is the target observee's identity store: stable identity markers that distinguish this entity from others and persist across interactions. Behavior, tendencies, transient state, and episodic facts belong in observations, not on the peer card.
|
||||
|
||||
A peer can be anything with identity that changes over time — a human, an agent, a codebase, a team, an organization. Do not assume the target observee is human. Do not require any field; empty is the correct output when evidence is absent.
|
||||
|
||||
### Allowed entry kinds
|
||||
|
||||
Each entry must start with one of these four prefixes (exact case, followed by a space):
|
||||
|
||||
- `IDENTITY: ...` — canonical name, kind, aliases, IDs
|
||||
- `IDENTITY: Name: Alice`
|
||||
- `IDENTITY: Kind: Python monorepo`
|
||||
- `IDENTITY: Version: 4.2`
|
||||
- `IDENTITY: Aliases: alice@example.com`
|
||||
- `ATTRIBUTE: ...` — stable durable property of the entity (including explicitly stated standing preferences)
|
||||
- `ATTRIBUTE: Location: NYC`
|
||||
- `ATTRIBUTE: Language: Python`
|
||||
- `ATTRIBUTE: Prefers tea`
|
||||
- `ATTRIBUTE: Charter: ship Honcho infrastructure`
|
||||
- `RELATIONSHIP: ...` — durable link to another entity
|
||||
- `RELATIONSHIP: Spouse: Bob`
|
||||
- `RELATIONSHIP: Maintainer: vineeth`
|
||||
- `RELATIONSHIP: Members: vineeth, rajat`
|
||||
- `INSTRUCTION: ...` — standing rule of engagement that the target observee has explicitly stated (do/don't for the observer). Only when explicit; never inferred from behavior.
|
||||
- `INSTRUCTION: Call me Vee`
|
||||
- `INSTRUCTION: Never push to main without review`
|
||||
|
||||
### Rules
|
||||
|
||||
1. **Stable.** If the value plausibly changes within six months absent a deliberate announcement, it does not belong on the card. Prefer leaving the card empty over filling it with volatile content.
|
||||
2. **Subject is the target observee.** Every entry must be a fact about the target observee, not about another participant in the session. Never write facts about co-occurring peers into the card, no matter how frequently they appear in the messages.
|
||||
3. **Evidence-grounded.** Only write what the target observee has explicitly stated, or what another participant has explicitly stated about the target observee with the target observee's assent. No "general knowledge" inferences (`"co-founder"` does not imply an age; mentioning a colleague does not imply a family relationship).
|
||||
4. **Type-agnostic.** The target observee may not be human. Do not require name/age/location/family/occupation fields.
|
||||
5. **No behavioral content.** TRAITs, behavioral tendencies, patterns, and inferred preferences belong in observations, not on the peer card. Do not write `TRAIT:` entries or behavioral `PREFERENCE:` entries — they will be rejected.
|
||||
6. **No evidence bundles.** Each entry is one concise fact. No `e.g.` clauses, no parenthetical example lists, no semicolon-separated value dumps.
|
||||
|
||||
### Migrating an existing peer card
|
||||
|
||||
The CURRENT PEER CARD shown in the user message may contain entries from an older format that do not start with an allowed prefix (e.g. `Name: Alice`, `Lives in NYC`, `TRAIT: Analytical`, `PREFERENCE: Detailed explanations`). When you call `update_peer_card`, you are responsible for re-emitting the entries you want to keep — entries you omit are dropped, and entries without an allowed prefix are silently rejected.
|
||||
|
||||
For each legacy entry:
|
||||
|
||||
- If it is still a valid identity marker, re-emit it under the correct prefix and keep the original content where reasonable. Examples:
|
||||
- `Name: Alice` → `IDENTITY: Name: Alice`
|
||||
- `Lives in NYC` → `ATTRIBUTE: Location: NYC`
|
||||
- `Works at Google` → `ATTRIBUTE: Employer: Google`
|
||||
- `INSTRUCTION: Call me Vee` → keep as is (already correctly prefixed)
|
||||
- Drop entries that violate the rules above: behavioral `TRAIT:` lines, inferred behavioral `PREFERENCE:` lines, one-off events, transient state, evidence bundles. Do not re-prefix them — they are not identity markers.
|
||||
|
||||
When in doubt about a specific legacy entry, prefer migrating it (so valid info isn't lost) over dropping it. Splitting one dense legacy entry into multiple correctly-prefixed entries is fine and encouraged (e.g. a semicolon-separated `Tech Stack:` dump can become several `ATTRIBUTE:` lines, one per durable tool/platform).
|
||||
|
||||
Call `update_peer_card` with the complete deduplicated list when there is a durable identity update to record, or when the existing card needs migration. Entries that do not start with one of the four allowed prefixes will be rejected. Keep concise (max 40 entries)."""
|
||||
|
||||
|
||||
class BaseSpecialist(ABC):
|
||||
"""Base class for agentic specialists."""
|
||||
|
|
@ -78,6 +137,10 @@ class BaseSpecialist(ABC):
|
|||
# Whether this specialist is allowed to write to the peer card. Defaults to True;
|
||||
# specialists that should never touch the card (e.g., induction) override to False.
|
||||
can_update_peer_card: bool = True
|
||||
# Whether the current peer card is fetched and injected into the user prompt.
|
||||
# Card-refresh runs in rebuild mode set this to False so the card is
|
||||
# reconstructed solely from observations present in the collection.
|
||||
inject_peer_card: bool = True
|
||||
# Subclasses can override to customize the peer card update instruction
|
||||
peer_card_update_instruction: str = (
|
||||
"Only update this with durable identity markers via `update_peer_card`."
|
||||
|
|
@ -218,9 +281,11 @@ If you update it, send the full deduplicated list and remove stale entries.
|
|||
configuration is None or configuration.peer_card.create
|
||||
)
|
||||
|
||||
# Fetch current peer card to inject into prompt (saves a tool call)
|
||||
# Fetch current peer card to inject into prompt (saves a tool call).
|
||||
# Skipped when inject_peer_card is False (card-refresh rebuild
|
||||
# mode): the card must be reconstructed from observations only.
|
||||
current_peer_card: list[str] | None = None
|
||||
if peer_card_enabled:
|
||||
if peer_card_enabled and self.inject_peer_card:
|
||||
current_peer_card = await crud.get_peer_card(
|
||||
db,
|
||||
workspace_name=workspace_name,
|
||||
|
|
@ -490,61 +555,7 @@ class DeductionSpecialist(BaseSpecialist):
|
|||
_ = observed
|
||||
peer_card_section = ""
|
||||
if peer_card_enabled:
|
||||
peer_card_section = """
|
||||
|
||||
## PEER CARD (REQUIRED)
|
||||
|
||||
The peer card is the target observee's identity store: stable identity markers that distinguish this entity from others and persist across interactions. Behavior, tendencies, transient state, and episodic facts belong in observations, not on the peer card.
|
||||
|
||||
A peer can be anything with identity that changes over time — a human, an agent, a codebase, a team, an organization. Do not assume the target observee is human. Do not require any field; empty is the correct output when evidence is absent.
|
||||
|
||||
### Allowed entry kinds
|
||||
|
||||
Each entry must start with one of these four prefixes (exact case, followed by a space):
|
||||
|
||||
- `IDENTITY: ...` — canonical name, kind, aliases, IDs
|
||||
- `IDENTITY: Name: Alice`
|
||||
- `IDENTITY: Kind: Python monorepo`
|
||||
- `IDENTITY: Version: 4.2`
|
||||
- `IDENTITY: Aliases: alice@example.com`
|
||||
- `ATTRIBUTE: ...` — stable durable property of the entity (including explicitly stated standing preferences)
|
||||
- `ATTRIBUTE: Location: NYC`
|
||||
- `ATTRIBUTE: Language: Python`
|
||||
- `ATTRIBUTE: Prefers tea`
|
||||
- `ATTRIBUTE: Charter: ship Honcho infrastructure`
|
||||
- `RELATIONSHIP: ...` — durable link to another entity
|
||||
- `RELATIONSHIP: Spouse: Bob`
|
||||
- `RELATIONSHIP: Maintainer: vineeth`
|
||||
- `RELATIONSHIP: Members: vineeth, rajat`
|
||||
- `INSTRUCTION: ...` — standing rule of engagement that the target observee has explicitly stated (do/don't for the observer). Only when explicit; never inferred from behavior.
|
||||
- `INSTRUCTION: Call me Vee`
|
||||
- `INSTRUCTION: Never push to main without review`
|
||||
|
||||
### Rules
|
||||
|
||||
1. **Stable.** If the value plausibly changes within six months absent a deliberate announcement, it does not belong on the card. Prefer leaving the card empty over filling it with volatile content.
|
||||
2. **Subject is the target observee.** Every entry must be a fact about the target observee, not about another participant in the session. Never write facts about co-occurring peers into the card, no matter how frequently they appear in the messages.
|
||||
3. **Evidence-grounded.** Only write what the target observee has explicitly stated, or what another participant has explicitly stated about the target observee with the target observee's assent. No "general knowledge" inferences (`"co-founder"` does not imply an age; mentioning a colleague does not imply a family relationship).
|
||||
4. **Type-agnostic.** The target observee may not be human. Do not require name/age/location/family/occupation fields.
|
||||
5. **No behavioral content.** TRAITs, behavioral tendencies, patterns, and inferred preferences belong in observations, not on the peer card. Do not write `TRAIT:` entries or behavioral `PREFERENCE:` entries — they will be rejected.
|
||||
6. **No evidence bundles.** Each entry is one concise fact. No `e.g.` clauses, no parenthetical example lists, no semicolon-separated value dumps.
|
||||
|
||||
### Migrating an existing peer card
|
||||
|
||||
The CURRENT PEER CARD shown in the user message may contain entries from an older format that do not start with an allowed prefix (e.g. `Name: Alice`, `Lives in NYC`, `TRAIT: Analytical`, `PREFERENCE: Detailed explanations`). When you call `update_peer_card`, you are responsible for re-emitting the entries you want to keep — entries you omit are dropped, and entries without an allowed prefix are silently rejected.
|
||||
|
||||
For each legacy entry:
|
||||
|
||||
- If it is still a valid identity marker, re-emit it under the correct prefix and keep the original content where reasonable. Examples:
|
||||
- `Name: Alice` → `IDENTITY: Name: Alice`
|
||||
- `Lives in NYC` → `ATTRIBUTE: Location: NYC`
|
||||
- `Works at Google` → `ATTRIBUTE: Employer: Google`
|
||||
- `INSTRUCTION: Call me Vee` → keep as is (already correctly prefixed)
|
||||
- Drop entries that violate the rules above: behavioral `TRAIT:` lines, inferred behavioral `PREFERENCE:` lines, one-off events, transient state, evidence bundles. Do not re-prefix them — they are not identity markers.
|
||||
|
||||
When in doubt about a specific legacy entry, prefer migrating it (so valid info isn't lost) over dropping it. Splitting one dense legacy entry into multiple correctly-prefixed entries is fine and encouraged (e.g. a semicolon-separated `Tech Stack:` dump can become several `ATTRIBUTE:` lines, one per durable tool/platform).
|
||||
|
||||
Call `update_peer_card` with the complete deduplicated list when there is a durable identity update to record, or when the existing card needs migration. Entries that do not start with one of the four allowed prefixes will be rejected. Keep concise (max 40 entries)."""
|
||||
peer_card_section = PEER_CARD_SYSTEM_SECTION
|
||||
|
||||
return f"""You are a deductive reasoning agent analyzing observations about the target observee.
|
||||
|
||||
|
|
@ -765,6 +776,113 @@ Remember: patterns need 2+ sources. Look for tendencies, preferences, and behavi
|
|||
Go."""
|
||||
|
||||
|
||||
class CardRefreshSpecialist(BaseSpecialist):
|
||||
"""
|
||||
Card-only maintenance specialist for the ``card_refresh`` dream type.
|
||||
|
||||
Restricted to peer-card work: it may discover observations
|
||||
(get_recent_observations, search_memory) and rewrite the peer card
|
||||
(update_peer_card). It has NO observation-mutating tools — a card refresh
|
||||
must never create or delete observations.
|
||||
|
||||
Two modes:
|
||||
- refresh (default): the current card is injected into the prompt and the
|
||||
specialist folds in new identity markers.
|
||||
- rebuild: the current card is NOT injected; the specialist reconstructs
|
||||
the card solely from observations present in the collection. Used after
|
||||
removals, where the old card may contain facts whose support was deleted.
|
||||
|
||||
Not a singleton — instantiated per run because ``rebuild`` is per-dream
|
||||
state.
|
||||
"""
|
||||
|
||||
name: str = "card_refresh"
|
||||
peer_card_update_instruction: str = "Update this with `update_peer_card`. See the PEER CARD section in the system prompt for the allowed entry kinds and rules."
|
||||
|
||||
# Low iteration ceiling for this lightweight, single-purpose run.
|
||||
MAX_ITERATIONS_CEILING: int = 6
|
||||
|
||||
def __init__(self, *, rebuild: bool = False) -> None:
|
||||
self.rebuild: bool = rebuild
|
||||
# In rebuild mode the existing card is withheld from the prompt.
|
||||
self.inject_peer_card: bool = not rebuild
|
||||
|
||||
def get_tools(self, *, peer_card_enabled: bool = True) -> list[dict[str, Any]]:
|
||||
if peer_card_enabled:
|
||||
return CARD_REFRESH_SPECIALIST_TOOLS
|
||||
# Defensive: a card refresh without card write access is a no-op, and
|
||||
# the orchestrator skips the run entirely when peer cards are disabled.
|
||||
return [
|
||||
t
|
||||
for t in CARD_REFRESH_SPECIALIST_TOOLS
|
||||
if t["name"] not in PEER_CARD_TOOL_NAMES
|
||||
]
|
||||
|
||||
def get_model_config(self) -> ConfiguredModelSettings:
|
||||
# Card refresh is a deduction-family task; reuse its model config.
|
||||
return _require_specialist_model_config(
|
||||
settings.DREAM.DEDUCTION_MODEL_CONFIG,
|
||||
specialist_name="DREAM CARD_REFRESH",
|
||||
)
|
||||
|
||||
def get_max_tokens(self) -> int:
|
||||
return 8192
|
||||
|
||||
def get_max_iterations(self) -> int:
|
||||
return min(self.MAX_ITERATIONS_CEILING, settings.DREAM.MAX_TOOL_ITERATIONS)
|
||||
|
||||
def build_system_prompt(
|
||||
self, observed: str, *, peer_card_enabled: bool = True
|
||||
) -> str:
|
||||
_ = observed
|
||||
_ = peer_card_enabled
|
||||
rebuild_section = ""
|
||||
if self.rebuild:
|
||||
rebuild_section = """
|
||||
|
||||
## REBUILD MODE
|
||||
|
||||
The existing peer card is deliberately NOT shown to you: it may contain entries whose supporting observations have since been removed. Build the card solely from the observations you find in the collection right now. Do not carry over or guess at prior card content — if an identity marker is not supported by a current observation, it does not go on the card."""
|
||||
|
||||
return f"""You are a peer-card maintenance agent for the target observee.
|
||||
|
||||
## YOUR JOB
|
||||
|
||||
Refresh the peer card and nothing else. You cannot create or delete observations — you have no tools for that. Your only write operation is `update_peer_card`.
|
||||
|
||||
## PROCESS
|
||||
|
||||
1. Survey the observation space: start with `get_recent_observations`, then use `search_memory` for targeted follow-ups (names, roles, locations, standing instructions).
|
||||
2. Extract stable identity markers supported by the observations you found.
|
||||
3. Call `update_peer_card` once with the complete deduplicated list.
|
||||
|
||||
Keep it short — a handful of tool calls at most.{rebuild_section}
|
||||
{PEER_CARD_SYSTEM_SECTION}"""
|
||||
|
||||
def build_user_prompt(
|
||||
self,
|
||||
observed: str,
|
||||
hints: list[str] | None,
|
||||
peer_card: list[str] | None = None,
|
||||
) -> str:
|
||||
_ = hints
|
||||
target_observee_context = self._build_target_observee_context(observed)
|
||||
peer_card_context = self._build_peer_card_context(peer_card)
|
||||
|
||||
if self.rebuild:
|
||||
return f"""{target_observee_context}Rebuild the peer card from scratch.
|
||||
|
||||
The previous card is not shown and must not be assumed: reconstruct the card solely from observations currently in the collection. Start with `get_recent_observations`, verify with `search_memory` where needed, then call `update_peer_card` with the complete list.
|
||||
|
||||
Go."""
|
||||
|
||||
return f"""{target_observee_context}{peer_card_context}Refresh the peer card.
|
||||
|
||||
Review recent observations with `get_recent_observations` (and `search_memory` for targeted checks), then call `update_peer_card` with the complete deduplicated list if there is anything to add, correct, or migrate. If the card is already accurate and complete, finish without updating it.
|
||||
|
||||
Go."""
|
||||
|
||||
|
||||
# Singleton instances
|
||||
SPECIALISTS: dict[str, BaseSpecialist] = {
|
||||
"deduction": DeductionSpecialist(),
|
||||
|
|
|
|||
|
|
@ -231,6 +231,7 @@ async def schedule_dream(
|
|||
observed=observed,
|
||||
dream_type=dream_type,
|
||||
session_name=request.session_id,
|
||||
rebuild=request.rebuild,
|
||||
# Manual route — explicit sentinels for the DreamRunEvent
|
||||
# scheduling-context fields. Auto-schedule threads concrete
|
||||
# threshold/delay reasons (see src/dreamer/dream_scheduler.py);
|
||||
|
|
|
|||
|
|
@ -670,6 +670,14 @@ class ScheduleDreamRequest(BaseModel):
|
|||
session_id: str | None = Field(
|
||||
None, description="Session ID to scope the dream to if specified"
|
||||
)
|
||||
rebuild: bool = Field(
|
||||
False,
|
||||
description=(
|
||||
"card_refresh dreams only: rebuild the peer card solely from "
|
||||
"observations currently in the collection, without injecting the "
|
||||
"existing card (use after removals)"
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
|
|
|
|||
|
|
@ -17,6 +17,10 @@ class DreamType(str, Enum):
|
|||
"""Types of dreams that can be triggered."""
|
||||
|
||||
OMNI = "omni"
|
||||
# Lightweight card-only refresh: runs a single specialist restricted to
|
||||
# peer-card tools. Used for event-driven refreshes (scope membership
|
||||
# changes, cold starts) — never creates or deletes observations.
|
||||
CARD_REFRESH = "card_refresh"
|
||||
|
||||
|
||||
class ReasoningConfiguration(BaseModel):
|
||||
|
|
|
|||
|
|
@ -34,6 +34,7 @@ class CallPurpose(str, Enum):
|
|||
DIALECTIC_ANSWER = "dialectic.answer"
|
||||
DREAM_DEDUCTION = "dream.deduction"
|
||||
DREAM_INDUCTION = "dream.induction"
|
||||
DREAM_CARD_REFRESH = "dream.card_refresh"
|
||||
SUMMARY_SHORT = "summary.short"
|
||||
SUMMARY_LONG = "summary.long"
|
||||
|
||||
|
|
|
|||
|
|
@ -850,6 +850,18 @@ INDUCTION_SPECIALIST_TOOLS: list[dict[str, Any]] = [
|
|||
TOOLS["create_observations_inductive"],
|
||||
]
|
||||
|
||||
# Tools for the card-refresh specialist (card_refresh dream type).
|
||||
# Card-only maintenance: discovery plus update_peer_card. Deliberately
|
||||
# excludes every observation-mutating tool (create_observations*,
|
||||
# delete_observations) — a card refresh must never touch observations.
|
||||
CARD_REFRESH_SPECIALIST_TOOLS: list[dict[str, Any]] = [
|
||||
# Discovery tools
|
||||
TOOLS["get_recent_observations"],
|
||||
TOOLS["search_memory"],
|
||||
# Action tool
|
||||
TOOLS["update_peer_card"],
|
||||
]
|
||||
|
||||
|
||||
async def create_observations(
|
||||
observations: list[schemas.ObservationInput],
|
||||
|
|
@ -1410,6 +1422,21 @@ async def _handle_create_observations_impl(
|
|||
)
|
||||
)
|
||||
continue
|
||||
# Session-purity invariant: explicit observations record what was
|
||||
# directly derived from a session's messages. Agents that are not
|
||||
# processing messages (dreamer specialists, dialectic) must not mint
|
||||
# them — consolidation output belongs at a derived level.
|
||||
if not ctx.current_messages and validated.level == "explicit":
|
||||
validation_failures.append(
|
||||
ObservationFailure(
|
||||
content_preview=validated.content[:50],
|
||||
error=(
|
||||
"Only message ingestion can create 'explicit' observations; "
|
||||
"use a derived level (deductive/inductive/contradiction)"
|
||||
),
|
||||
)
|
||||
)
|
||||
continue
|
||||
observations.append(validated)
|
||||
|
||||
if not observations:
|
||||
|
|
|
|||
|
|
@ -66,6 +66,11 @@ class DreamPayload(BasePayload):
|
|||
delay_reason: str | None = None
|
||||
documents_since_last_dream_at_schedule: int | None = None
|
||||
document_threshold: int | None = None
|
||||
# card_refresh only: when True the existing peer card is NOT injected into
|
||||
# the specialist prompt and the card is rebuilt solely from observations
|
||||
# currently in the collection (used after removals, where the old card may
|
||||
# contain facts whose support was deleted).
|
||||
rebuild: bool = False
|
||||
|
||||
|
||||
class DeletionPayload(BasePayload):
|
||||
|
|
@ -103,6 +108,7 @@ def create_dream_payload(
|
|||
delay_reason: str | None = None,
|
||||
documents_since_last_dream_at_schedule: int | None = None,
|
||||
document_threshold: int | None = None,
|
||||
rebuild: bool = False,
|
||||
) -> dict[str, Any]:
|
||||
"""Create a dream payload."""
|
||||
return DreamPayload(
|
||||
|
|
@ -114,6 +120,7 @@ def create_dream_payload(
|
|||
delay_reason=delay_reason,
|
||||
documents_since_last_dream_at_schedule=documents_since_last_dream_at_schedule,
|
||||
document_threshold=document_threshold,
|
||||
rebuild=rebuild,
|
||||
).model_dump(mode="json", exclude_none=True)
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -1,4 +1,5 @@
|
|||
import datetime
|
||||
from unittest.mock import AsyncMock, patch
|
||||
|
||||
import pytest
|
||||
from nanoid import generate as generate_nanoid
|
||||
|
|
@ -1004,3 +1005,332 @@ class TestDocumentCRUD:
|
|||
assert len(documents) == 2
|
||||
assert documents[0].content in ["Observation 1", "Observation 2"]
|
||||
assert documents[1].content in ["Observation 1", "Observation 2"]
|
||||
|
||||
|
||||
class TestSessionPurityInvariant:
|
||||
"""Regression tests for the explicit-document session-purity invariant.
|
||||
|
||||
Explicit documents are session-pure records of what was derived from one
|
||||
session's messages (the Scopes copy-by-session model depends on this):
|
||||
|
||||
- an explicit document must always carry a non-null session_name
|
||||
- dedup/merge (exact and semantic) must never cross document levels
|
||||
- dedup/merge must never cross sessions for explicit documents
|
||||
"""
|
||||
|
||||
async def _setup(
|
||||
self,
|
||||
db_session: AsyncSession,
|
||||
test_workspace: models.Workspace,
|
||||
test_peer: models.Peer,
|
||||
) -> tuple[models.Peer, models.Session, models.Session]:
|
||||
"""Create an observed peer, two sessions, and the collection."""
|
||||
test_peer2 = models.Peer(
|
||||
name=str(generate_nanoid()), workspace_name=test_workspace.name
|
||||
)
|
||||
db_session.add(test_peer2)
|
||||
session_a = models.Session(
|
||||
name=str(generate_nanoid()), workspace_name=test_workspace.name
|
||||
)
|
||||
session_b = models.Session(
|
||||
name=str(generate_nanoid()), workspace_name=test_workspace.name
|
||||
)
|
||||
db_session.add_all([session_a, session_b])
|
||||
await db_session.flush()
|
||||
|
||||
collection = models.Collection(
|
||||
workspace_name=test_workspace.name,
|
||||
observer=test_peer.name,
|
||||
observed=test_peer2.name,
|
||||
)
|
||||
db_session.add(collection)
|
||||
await db_session.flush()
|
||||
return test_peer2, session_a, session_b
|
||||
|
||||
def _doc(
|
||||
self,
|
||||
content: str,
|
||||
*,
|
||||
session_name: str | None,
|
||||
level: str = "explicit",
|
||||
message_id: int = 1,
|
||||
) -> schemas.DocumentCreate:
|
||||
return schemas.DocumentCreate(
|
||||
content=content,
|
||||
embedding=[0.1] * 1536,
|
||||
session_name=session_name,
|
||||
level=level, # pyright: ignore[reportArgumentType]
|
||||
metadata=schemas.DocumentMetadata(
|
||||
message_ids=[message_id],
|
||||
message_created_at="2026-01-01T00:00:00Z",
|
||||
),
|
||||
)
|
||||
|
||||
async def _live_docs(
|
||||
self,
|
||||
db_session: AsyncSession,
|
||||
workspace_name: str,
|
||||
observer: str,
|
||||
observed: str,
|
||||
) -> list[models.Document]:
|
||||
return list(
|
||||
(
|
||||
await db_session.execute(
|
||||
select(models.Document).where(
|
||||
models.Document.workspace_name == workspace_name,
|
||||
models.Document.observer == observer,
|
||||
models.Document.observed == observed,
|
||||
models.Document.deleted_at.is_(None),
|
||||
)
|
||||
)
|
||||
)
|
||||
.scalars()
|
||||
.all()
|
||||
)
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_explicit_without_session_is_refused(
|
||||
self,
|
||||
db_session: AsyncSession,
|
||||
sample_data: tuple[models.Workspace, models.Peer],
|
||||
):
|
||||
"""An explicit document with session_name=None must not be written;
|
||||
derived levels remain allowed without a session (dream output)."""
|
||||
test_workspace, test_peer = sample_data
|
||||
test_peer2, _, _ = await self._setup(db_session, test_workspace, test_peer)
|
||||
|
||||
accepted = (
|
||||
await crud.create_documents(
|
||||
db_session,
|
||||
[
|
||||
self._doc("Global explicit fact", session_name=None),
|
||||
self._doc(
|
||||
"Dream-derived conclusion",
|
||||
session_name=None,
|
||||
level="deductive",
|
||||
message_id=2,
|
||||
),
|
||||
],
|
||||
workspace_name=test_workspace.name,
|
||||
observer=test_peer.name,
|
||||
observed=test_peer2.name,
|
||||
)
|
||||
).created_documents
|
||||
|
||||
assert [d.content for d in accepted] == ["Dream-derived conclusion"]
|
||||
live = await self._live_docs(
|
||||
db_session, test_workspace.name, test_peer.name, test_peer2.name
|
||||
)
|
||||
assert len(live) == 1
|
||||
assert live[0].level == "deductive"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_exact_dedup_never_merges_explicit_across_sessions(
|
||||
self,
|
||||
db_session: AsyncSession,
|
||||
sample_data: tuple[models.Workspace, models.Peer],
|
||||
):
|
||||
"""The same explicit fact stated in two sessions produces two
|
||||
session-pure documents; the other session's row is not reinforced."""
|
||||
test_workspace, test_peer = sample_data
|
||||
test_peer2, session_a, session_b = await self._setup(
|
||||
db_session, test_workspace, test_peer
|
||||
)
|
||||
|
||||
await crud.create_documents(
|
||||
db_session,
|
||||
[self._doc("User likes coffee", session_name=session_a.name)],
|
||||
workspace_name=test_workspace.name,
|
||||
observer=test_peer.name,
|
||||
observed=test_peer2.name,
|
||||
)
|
||||
accepted = (
|
||||
await crud.create_documents(
|
||||
db_session,
|
||||
[
|
||||
self._doc(
|
||||
"user likes coffee ", session_name=session_b.name, message_id=2
|
||||
)
|
||||
],
|
||||
workspace_name=test_workspace.name,
|
||||
observer=test_peer.name,
|
||||
observed=test_peer2.name,
|
||||
)
|
||||
).created_documents
|
||||
|
||||
assert len(accepted) == 1
|
||||
live = await self._live_docs(
|
||||
db_session, test_workspace.name, test_peer.name, test_peer2.name
|
||||
)
|
||||
assert len(live) == 2
|
||||
assert {doc.session_name for doc in live} == {session_a.name, session_b.name}
|
||||
assert all(doc.times_derived == 1 for doc in live)
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_exact_dedup_never_merges_across_levels(
|
||||
self,
|
||||
db_session: AsyncSession,
|
||||
sample_data: tuple[models.Workspace, models.Peer],
|
||||
):
|
||||
"""An explicit fact must not be dropped/reinforced against a derived
|
||||
document that happens to share its content."""
|
||||
test_workspace, test_peer = sample_data
|
||||
test_peer2, session_a, _ = await self._setup(
|
||||
db_session, test_workspace, test_peer
|
||||
)
|
||||
|
||||
await crud.create_documents(
|
||||
db_session,
|
||||
[
|
||||
self._doc(
|
||||
"User likes coffee", session_name=session_a.name, level="deductive"
|
||||
)
|
||||
],
|
||||
workspace_name=test_workspace.name,
|
||||
observer=test_peer.name,
|
||||
observed=test_peer2.name,
|
||||
)
|
||||
accepted = (
|
||||
await crud.create_documents(
|
||||
db_session,
|
||||
[
|
||||
self._doc(
|
||||
"User likes coffee", session_name=session_a.name, message_id=2
|
||||
)
|
||||
],
|
||||
workspace_name=test_workspace.name,
|
||||
observer=test_peer.name,
|
||||
observed=test_peer2.name,
|
||||
)
|
||||
).created_documents
|
||||
|
||||
assert len(accepted) == 1
|
||||
live = await self._live_docs(
|
||||
db_session, test_workspace.name, test_peer.name, test_peer2.name
|
||||
)
|
||||
assert len(live) == 2
|
||||
assert {doc.level for doc in live} == {"explicit", "deductive"}
|
||||
assert all(doc.times_derived == 1 for doc in live)
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_exact_dedup_still_merges_derived_levels_across_sessions(
|
||||
self,
|
||||
db_session: AsyncSession,
|
||||
sample_data: tuple[models.Workspace, models.Peer],
|
||||
):
|
||||
"""Derived levels are consolidations, not session-pure records:
|
||||
cross-session exact dedup still reinforces the existing row."""
|
||||
test_workspace, test_peer = sample_data
|
||||
test_peer2, session_a, session_b = await self._setup(
|
||||
db_session, test_workspace, test_peer
|
||||
)
|
||||
|
||||
await crud.create_documents(
|
||||
db_session,
|
||||
[
|
||||
self._doc(
|
||||
"Probably a morning person",
|
||||
session_name=session_a.name,
|
||||
level="deductive",
|
||||
)
|
||||
],
|
||||
workspace_name=test_workspace.name,
|
||||
observer=test_peer.name,
|
||||
observed=test_peer2.name,
|
||||
)
|
||||
accepted = (
|
||||
await crud.create_documents(
|
||||
db_session,
|
||||
[
|
||||
self._doc(
|
||||
"probably a morning person",
|
||||
session_name=session_b.name,
|
||||
level="deductive",
|
||||
message_id=2,
|
||||
)
|
||||
],
|
||||
workspace_name=test_workspace.name,
|
||||
observer=test_peer.name,
|
||||
observed=test_peer2.name,
|
||||
)
|
||||
).created_documents
|
||||
|
||||
assert len(accepted) == 0
|
||||
live = await self._live_docs(
|
||||
db_session, test_workspace.name, test_peer.name, test_peer2.name
|
||||
)
|
||||
assert len(live) == 1
|
||||
assert live[0].times_derived == 2
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_semantic_dedup_scoped_to_level_and_session(
|
||||
self,
|
||||
db_session: AsyncSession,
|
||||
sample_data: tuple[models.Workspace, models.Peer],
|
||||
):
|
||||
"""is_rejected_duplicate must constrain candidate search to the same
|
||||
level, and to the same session for explicit documents."""
|
||||
test_workspace, test_peer = sample_data
|
||||
test_peer2, session_a, _ = await self._setup(
|
||||
db_session, test_workspace, test_peer
|
||||
)
|
||||
|
||||
explicit_doc = self._doc("User likes coffee", session_name=session_a.name)
|
||||
with patch(
|
||||
"src.crud.document.query_documents", new=AsyncMock(return_value=[])
|
||||
) as mock_query:
|
||||
rejected = await is_rejected_duplicate(
|
||||
db_session,
|
||||
explicit_doc,
|
||||
test_workspace.name,
|
||||
observer=test_peer.name,
|
||||
observed=test_peer2.name,
|
||||
)
|
||||
assert rejected is SemanticRejectionResult.NOT_DUPLICATE
|
||||
assert mock_query.await_args is not None
|
||||
assert mock_query.await_args.kwargs["filters"] == {
|
||||
"level": "explicit",
|
||||
"session_name": session_a.name,
|
||||
}
|
||||
|
||||
deductive_doc = self._doc(
|
||||
"User likes coffee", session_name=None, level="deductive"
|
||||
)
|
||||
with patch(
|
||||
"src.crud.document.query_documents", new=AsyncMock(return_value=[])
|
||||
) as mock_query:
|
||||
rejected = await is_rejected_duplicate(
|
||||
db_session,
|
||||
deductive_doc,
|
||||
test_workspace.name,
|
||||
observer=test_peer.name,
|
||||
observed=test_peer2.name,
|
||||
)
|
||||
assert rejected is SemanticRejectionResult.NOT_DUPLICATE
|
||||
assert mock_query.await_args is not None
|
||||
assert mock_query.await_args.kwargs["filters"] == {"level": "deductive"}
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_semantic_dedup_refuses_sessionless_explicit(
|
||||
self,
|
||||
db_session: AsyncSession,
|
||||
sample_data: tuple[models.Workspace, models.Peer],
|
||||
):
|
||||
"""A session-less explicit document has no valid merge partner: it is
|
||||
never treated as a duplicate and no candidate search runs."""
|
||||
test_workspace, test_peer = sample_data
|
||||
test_peer2, _, _ = await self._setup(db_session, test_workspace, test_peer)
|
||||
|
||||
doc = self._doc("User likes coffee", session_name=None)
|
||||
with patch(
|
||||
"src.crud.document.query_documents", new=AsyncMock(return_value=[])
|
||||
) as mock_query:
|
||||
rejected = await is_rejected_duplicate(
|
||||
db_session,
|
||||
doc,
|
||||
test_workspace.name,
|
||||
observer=test_peer.name,
|
||||
observed=test_peer2.name,
|
||||
)
|
||||
assert rejected is SemanticRejectionResult.NOT_DUPLICATE
|
||||
mock_query.assert_not_awaited()
|
||||
|
|
|
|||
|
|
@ -0,0 +1,335 @@
|
|||
"""Tests for the card_refresh dream type (DEV-2000, Scopes RFC prerequisite).
|
||||
|
||||
Covers:
|
||||
- queue plumbing: payload roundtrip, work-unit key isolation from omni,
|
||||
enqueue alongside a pending omni dream
|
||||
- process_dream dispatch of DreamType.CARD_REFRESH (and that it does NOT
|
||||
advance the omni dream guard pair)
|
||||
- specialist tool restriction (no observation-mutating tools)
|
||||
- the low tool-iteration cap
|
||||
- rebuild mode omitting the prior peer card from the prompt
|
||||
"""
|
||||
|
||||
from typing import Any
|
||||
from unittest.mock import AsyncMock, patch
|
||||
|
||||
import pytest
|
||||
import pytest_asyncio
|
||||
from sqlalchemy import select
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from src import models
|
||||
from src.config import settings
|
||||
from src.deriver.enqueue import enqueue_dream
|
||||
from src.dreamer.orchestrator import DreamResult, process_dream
|
||||
from src.dreamer.specialists import CardRefreshSpecialist
|
||||
from src.llm import HonchoLLMCallResponse
|
||||
from src.schemas import DreamType
|
||||
from src.utils.queue_payload import DreamPayload, create_dream_payload
|
||||
from src.utils.work_unit import construct_work_unit_key, parse_work_unit_key
|
||||
|
||||
OBSERVATION_MUTATION_TOOLS = {
|
||||
"create_observations",
|
||||
"create_observations_deductive",
|
||||
"create_observations_inductive",
|
||||
"delete_observations",
|
||||
}
|
||||
|
||||
|
||||
@pytest_asyncio.fixture
|
||||
async def seeded_collection(
|
||||
db_session: AsyncSession,
|
||||
sample_data: tuple[models.Workspace, models.Peer],
|
||||
) -> models.Collection:
|
||||
"""Create a Collection with an empty dream metadata dict."""
|
||||
workspace, peer = sample_data
|
||||
collection = models.Collection(
|
||||
observer=peer.name,
|
||||
observed=peer.name,
|
||||
workspace_name=workspace.name,
|
||||
internal_metadata={},
|
||||
)
|
||||
db_session.add(collection)
|
||||
await db_session.commit()
|
||||
await db_session.refresh(collection)
|
||||
return collection
|
||||
|
||||
|
||||
def _make_card_refresh_result() -> DreamResult:
|
||||
return DreamResult(
|
||||
run_id="test_run_card",
|
||||
specialists_run=["card_refresh"],
|
||||
deduction_success=True,
|
||||
induction_success=False,
|
||||
surprisal_enabled=False,
|
||||
surprisal_conclusion_count=0,
|
||||
total_iterations=2,
|
||||
total_duration_ms=42.0,
|
||||
input_tokens=10,
|
||||
output_tokens=5,
|
||||
)
|
||||
|
||||
|
||||
class TestQueuePlumbing:
|
||||
def test_payload_roundtrip_carries_rebuild(self):
|
||||
payload_dict = create_dream_payload(
|
||||
DreamType.CARD_REFRESH,
|
||||
observer="alice",
|
||||
observed="bob",
|
||||
rebuild=True,
|
||||
)
|
||||
validated = DreamPayload(**payload_dict)
|
||||
assert validated.dream_type == DreamType.CARD_REFRESH
|
||||
assert validated.rebuild is True
|
||||
|
||||
# Default is False, including for older payloads missing the field.
|
||||
assert (
|
||||
DreamPayload(dream_type=DreamType.OMNI, observer="a", observed="b").rebuild
|
||||
is False
|
||||
)
|
||||
|
||||
def test_work_unit_key_does_not_collide_with_omni(self):
|
||||
base = {"task_type": "dream", "observer": "alice", "observed": "bob"}
|
||||
omni_key = construct_work_unit_key("ws", {**base, "dream_type": "omni"})
|
||||
card_key = construct_work_unit_key("ws", {**base, "dream_type": "card_refresh"})
|
||||
|
||||
assert omni_key != card_key
|
||||
parsed = parse_work_unit_key(card_key)
|
||||
assert parsed.task_type == "dream"
|
||||
assert parsed.dream_type == "card_refresh"
|
||||
assert parsed.observer == "alice"
|
||||
assert parsed.observed == "bob"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_enqueue_alongside_pending_omni(
|
||||
self,
|
||||
db_session: AsyncSession,
|
||||
seeded_collection: models.Collection,
|
||||
):
|
||||
"""A pending omni dream must not dedupe away a card_refresh enqueue —
|
||||
the work-unit keys differ by dream type."""
|
||||
await enqueue_dream(
|
||||
seeded_collection.workspace_name,
|
||||
observer=seeded_collection.observer,
|
||||
observed=seeded_collection.observed,
|
||||
dream_type=DreamType.OMNI,
|
||||
)
|
||||
await enqueue_dream(
|
||||
seeded_collection.workspace_name,
|
||||
observer=seeded_collection.observer,
|
||||
observed=seeded_collection.observed,
|
||||
dream_type=DreamType.CARD_REFRESH,
|
||||
rebuild=True,
|
||||
)
|
||||
|
||||
items = (
|
||||
(
|
||||
await db_session.execute(
|
||||
select(models.QueueItem).where(
|
||||
models.QueueItem.workspace_name
|
||||
== seeded_collection.workspace_name,
|
||||
models.QueueItem.task_type == "dream",
|
||||
models.QueueItem.processed == False, # noqa: E712
|
||||
)
|
||||
)
|
||||
)
|
||||
.scalars()
|
||||
.all()
|
||||
)
|
||||
assert len(items) == 2
|
||||
dream_types = {item.payload["dream_type"] for item in items}
|
||||
assert dream_types == {"omni", "card_refresh"}
|
||||
card_item = next(
|
||||
item for item in items if item.payload["dream_type"] == "card_refresh"
|
||||
)
|
||||
assert card_item.payload["rebuild"] is True
|
||||
|
||||
|
||||
class TestProcessDreamDispatch:
|
||||
@pytest.mark.asyncio
|
||||
async def test_dispatches_card_refresh(
|
||||
self,
|
||||
seeded_collection: models.Collection,
|
||||
):
|
||||
payload = DreamPayload(
|
||||
dream_type=DreamType.CARD_REFRESH,
|
||||
observer=seeded_collection.observer,
|
||||
observed=seeded_collection.observed,
|
||||
rebuild=True,
|
||||
trigger_reason="manual",
|
||||
)
|
||||
|
||||
with patch(
|
||||
"src.dreamer.orchestrator.run_card_refresh_dream",
|
||||
new=AsyncMock(return_value=_make_card_refresh_result()),
|
||||
) as mock_run:
|
||||
await process_dream(payload, seeded_collection.workspace_name)
|
||||
|
||||
assert mock_run.await_args is not None
|
||||
kwargs = mock_run.await_args.kwargs
|
||||
assert kwargs["workspace_name"] == seeded_collection.workspace_name
|
||||
assert kwargs["observer"] == seeded_collection.observer
|
||||
assert kwargs["observed"] == seeded_collection.observed
|
||||
assert kwargs["rebuild"] is True
|
||||
assert kwargs["dream_type"] == "card_refresh"
|
||||
assert kwargs["trigger_reason"] == "manual"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_card_refresh_does_not_advance_dream_guard(
|
||||
self,
|
||||
db_session: AsyncSession,
|
||||
seeded_collection: models.Collection,
|
||||
):
|
||||
"""The omni guard pair (last_dream_at / last_dream_document_count)
|
||||
must not move on a card refresh — it would delay real consolidation."""
|
||||
payload = DreamPayload(
|
||||
dream_type=DreamType.CARD_REFRESH,
|
||||
observer=seeded_collection.observer,
|
||||
observed=seeded_collection.observed,
|
||||
)
|
||||
|
||||
with patch(
|
||||
"src.dreamer.orchestrator.run_card_refresh_dream",
|
||||
new=AsyncMock(return_value=_make_card_refresh_result()),
|
||||
):
|
||||
await process_dream(payload, seeded_collection.workspace_name)
|
||||
|
||||
await db_session.refresh(seeded_collection)
|
||||
dream_meta: dict[str, Any] = seeded_collection.internal_metadata.get(
|
||||
"dream", {}
|
||||
)
|
||||
assert "last_dream_at" not in dream_meta
|
||||
assert "last_dream_document_count" not in dream_meta
|
||||
|
||||
|
||||
class TestCardRefreshSpecialist:
|
||||
def test_tools_exclude_observation_mutation(self):
|
||||
for rebuild in (False, True):
|
||||
specialist = CardRefreshSpecialist(rebuild=rebuild)
|
||||
tool_names = {t["name"] for t in specialist.get_tools()}
|
||||
assert tool_names == {
|
||||
"get_recent_observations",
|
||||
"search_memory",
|
||||
"update_peer_card",
|
||||
}
|
||||
assert not tool_names & OBSERVATION_MUTATION_TOOLS
|
||||
|
||||
def test_tools_without_peer_card_strip_update(self):
|
||||
specialist = CardRefreshSpecialist()
|
||||
tool_names = {t["name"] for t in specialist.get_tools(peer_card_enabled=False)}
|
||||
assert "update_peer_card" not in tool_names
|
||||
assert not tool_names & OBSERVATION_MUTATION_TOOLS
|
||||
|
||||
def test_low_iteration_cap(self, monkeypatch: pytest.MonkeyPatch):
|
||||
specialist = CardRefreshSpecialist()
|
||||
assert specialist.get_max_iterations() == min(
|
||||
6, settings.DREAM.MAX_TOOL_ITERATIONS
|
||||
)
|
||||
|
||||
monkeypatch.setattr(settings.DREAM, "MAX_TOOL_ITERATIONS", 4)
|
||||
assert specialist.get_max_iterations() == 4
|
||||
|
||||
monkeypatch.setattr(settings.DREAM, "MAX_TOOL_ITERATIONS", 30)
|
||||
assert specialist.get_max_iterations() == 6
|
||||
|
||||
def test_rebuild_flag_controls_card_injection(self):
|
||||
assert CardRefreshSpecialist(rebuild=False).inject_peer_card is True
|
||||
assert CardRefreshSpecialist(rebuild=True).inject_peer_card is False
|
||||
|
||||
def test_rebuild_prompts_instruct_observation_only_build(self):
|
||||
specialist = CardRefreshSpecialist(rebuild=True)
|
||||
system_prompt = specialist.build_system_prompt("alice")
|
||||
assert "REBUILD MODE" in system_prompt
|
||||
assert "solely from the observations" in system_prompt
|
||||
|
||||
user_prompt = specialist.build_user_prompt("alice", hints=None, peer_card=None)
|
||||
assert "Rebuild the peer card" in user_prompt
|
||||
assert "CURRENT PEER CARD" not in user_prompt
|
||||
|
||||
async def _run_specialist(
|
||||
self, specialist: CardRefreshSpecialist, stored_card: list[str]
|
||||
) -> tuple[AsyncMock, AsyncMock]:
|
||||
"""Run the specialist with a fully mocked LLM layer; returns the
|
||||
(get_peer_card, honcho_llm_call) mocks for inspection."""
|
||||
mock_response = HonchoLLMCallResponse(
|
||||
content="done",
|
||||
input_tokens=10,
|
||||
output_tokens=5,
|
||||
finish_reasons=["stop"],
|
||||
)
|
||||
mock_get_peer_card = AsyncMock(return_value=stored_card)
|
||||
mock_llm_call = AsyncMock(return_value=mock_response)
|
||||
|
||||
with (
|
||||
patch("src.dreamer.specialists.crud.get_peer", new=AsyncMock()),
|
||||
patch(
|
||||
"src.dreamer.specialists.crud.get_peer_card",
|
||||
new=mock_get_peer_card,
|
||||
),
|
||||
patch(
|
||||
"src.dreamer.specialists.create_tool_executor",
|
||||
new=AsyncMock(return_value=AsyncMock()),
|
||||
),
|
||||
patch(
|
||||
"src.dreamer.specialists.honcho_llm_call",
|
||||
new=mock_llm_call,
|
||||
),
|
||||
):
|
||||
result = await specialist.run(
|
||||
workspace_name="workspace",
|
||||
observer="alice",
|
||||
observed="alice",
|
||||
session_name=None,
|
||||
)
|
||||
assert result.success is True
|
||||
return mock_get_peer_card, mock_llm_call
|
||||
|
||||
# Sentinel card entry that cannot collide with the prompt's own examples
|
||||
# (the shared PEER CARD section contains e.g. "IDENTITY: Name: Alice").
|
||||
STORED_CARD: list[str] = [
|
||||
"IDENTITY: Name: Zorblax-Prime",
|
||||
"ATTRIBUTE: Location: Ganymede",
|
||||
]
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_refresh_mode_injects_existing_card(
|
||||
self, monkeypatch: pytest.MonkeyPatch
|
||||
):
|
||||
monkeypatch.setattr(settings.METRICS, "ENABLED", False)
|
||||
|
||||
mock_get_peer_card, mock_llm_call = await self._run_specialist(
|
||||
CardRefreshSpecialist(rebuild=False), self.STORED_CARD
|
||||
)
|
||||
|
||||
mock_get_peer_card.assert_awaited_once()
|
||||
assert mock_llm_call.await_args is not None
|
||||
kwargs = mock_llm_call.await_args.kwargs
|
||||
user_message = kwargs["messages"][1]["content"]
|
||||
assert "IDENTITY: Name: Zorblax-Prime" in user_message
|
||||
assert "CURRENT PEER CARD" in user_message
|
||||
# Restricted tool offering and low iteration cap reach the LLM call.
|
||||
tool_names = {t["name"] for t in kwargs["tools"]}
|
||||
assert not tool_names & OBSERVATION_MUTATION_TOOLS
|
||||
assert kwargs["max_tool_iterations"] == min(
|
||||
6, settings.DREAM.MAX_TOOL_ITERATIONS
|
||||
)
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_rebuild_mode_omits_existing_card(
|
||||
self, monkeypatch: pytest.MonkeyPatch
|
||||
):
|
||||
monkeypatch.setattr(settings.METRICS, "ENABLED", False)
|
||||
|
||||
mock_get_peer_card, mock_llm_call = await self._run_specialist(
|
||||
CardRefreshSpecialist(rebuild=True), self.STORED_CARD
|
||||
)
|
||||
|
||||
# The stored card is never even fetched, let alone injected.
|
||||
mock_get_peer_card.assert_not_awaited()
|
||||
assert mock_llm_call.await_args is not None
|
||||
kwargs = mock_llm_call.await_args.kwargs
|
||||
for message in kwargs["messages"]:
|
||||
assert "IDENTITY: Name: Zorblax-Prime" not in message["content"]
|
||||
# No CURRENT PEER CARD block in the user prompt (the system prompt's
|
||||
# shared taxonomy section legitimately mentions the phrase).
|
||||
assert "CURRENT PEER CARD" not in kwargs["messages"][1]["content"]
|
||||
|
|
@ -9,6 +9,7 @@ from sqlalchemy.ext.asyncio import AsyncSession
|
|||
|
||||
from src import models
|
||||
from src.models import Peer, Workspace
|
||||
from src.schemas import DreamType
|
||||
|
||||
|
||||
def test_get_or_create_workspace(client: TestClient):
|
||||
|
|
@ -758,3 +759,52 @@ async def test_schedule_dream_invokes_enqueue_dream(
|
|||
"Loop 4: enqueue_dream no longer accepts document_count; the baseline "
|
||||
"is written atomically with last_dream_at in process_dream."
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_schedule_dream_card_refresh_forwards_rebuild(
|
||||
client: TestClient,
|
||||
db_session: AsyncSession,
|
||||
sample_data: tuple[Workspace, Peer],
|
||||
):
|
||||
"""POST /schedule_dream accepts dream_type=card_refresh and forwards the
|
||||
rebuild flag to enqueue_dream (manual/event-driven card refreshes bypass
|
||||
the volume gates by design)."""
|
||||
workspace, peer = sample_data
|
||||
|
||||
collection = models.Collection(
|
||||
observer=peer.name,
|
||||
observed=peer.name,
|
||||
workspace_name=workspace.name,
|
||||
internal_metadata={},
|
||||
)
|
||||
db_session.add(collection)
|
||||
await db_session.commit()
|
||||
|
||||
captured: dict[str, Any] = {}
|
||||
|
||||
async def fake_enqueue_dream(*args: Any, **kwargs: Any) -> None:
|
||||
captured["args"] = args
|
||||
captured["kwargs"] = kwargs
|
||||
|
||||
with (
|
||||
patch("src.routers.workspaces.settings.DREAM.ENABLED", True),
|
||||
patch(
|
||||
"src.routers.workspaces.enqueue_dream",
|
||||
new=AsyncMock(side_effect=fake_enqueue_dream),
|
||||
),
|
||||
):
|
||||
response = client.post(
|
||||
f"/v3/workspaces/{workspace.name}/schedule_dream",
|
||||
json={
|
||||
"observer": peer.name,
|
||||
"observed": peer.name,
|
||||
"dream_type": "card_refresh",
|
||||
"rebuild": True,
|
||||
},
|
||||
)
|
||||
|
||||
assert response.status_code == 204, response.text
|
||||
assert "kwargs" in captured, "enqueue_dream was not called"
|
||||
assert captured["kwargs"]["dream_type"] == DreamType.CARD_REFRESH
|
||||
assert captured["kwargs"]["rebuild"] is True
|
||||
|
|
|
|||
|
|
@ -239,6 +239,7 @@ class TestLLMCallCompletedEvent:
|
|||
assert CallPurpose.DIALECTIC_ANSWER.value == "dialectic.answer"
|
||||
assert CallPurpose.DREAM_DEDUCTION.value == "dream.deduction"
|
||||
assert CallPurpose.DREAM_INDUCTION.value == "dream.induction"
|
||||
assert CallPurpose.DREAM_CARD_REFRESH.value == "dream.card_refresh"
|
||||
assert CallPurpose.SUMMARY_SHORT.value == "summary.short"
|
||||
assert CallPurpose.SUMMARY_LONG.value == "summary.long"
|
||||
|
||||
|
|
|
|||
|
|
@ -248,6 +248,36 @@ class TestCreateObservations:
|
|||
assert doc.level == "deductive"
|
||||
assert doc.source_ids == ["premise1", "premise2"]
|
||||
|
||||
async def test_non_deriver_context_rejects_explicit(
|
||||
self,
|
||||
db_session: AsyncSession,
|
||||
make_tool_context: Callable[..., ToolContext],
|
||||
):
|
||||
"""Session-purity invariant: agents without current_messages (dreamer
|
||||
specialists, dialectic) must not create explicit-level observations,
|
||||
even when they pass level='explicit' to the generic tool."""
|
||||
ctx = make_tool_context(current_messages=None)
|
||||
|
||||
result = await _handle_create_observations(
|
||||
ctx,
|
||||
{
|
||||
"observations": [
|
||||
{"content": "Claims to be a doctor", "level": "explicit"},
|
||||
]
|
||||
},
|
||||
)
|
||||
|
||||
assert isinstance(result, str)
|
||||
assert "ERROR" in result
|
||||
assert "explicit" in result
|
||||
|
||||
# Verify nothing landed in the DB
|
||||
stmt = select(models.Document).where(
|
||||
models.Document.content == "Claims to be a doctor"
|
||||
)
|
||||
doc = (await db_session.execute(stmt)).scalar_one_or_none()
|
||||
assert doc is None
|
||||
|
||||
async def test_source_ids_display_prefix_is_stripped(
|
||||
self,
|
||||
db_session: AsyncSession,
|
||||
|
|
|
|||
Loading…
Reference in New Issue