mirror of
https://github.com/openswarm-ai/openswarm.git
synced 2026-08-20 11:42:23 +02:00
236 lines
9.1 KiB
Python
236 lines
9.1 KiB
Python
"""Pattern-miner invariants: evidence gathering excludes what must never count
|
|
(sub-agents, workflow runs, already-automated titles, stale/empty sessions),
|
|
cadence + counts are computed in code from verified evidence (never the aux
|
|
model's claims), dismissed patterns stay dismissed, the kill switch gates the
|
|
miner itself, and accept creates a real scheduled workflow.
|
|
|
|
Run:
|
|
cd backend && .venv/bin/python -m pytest tests/test_pattern_miner.py -v
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import json
|
|
from datetime import datetime, timedelta
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
|
|
|
|
def p_run(coro):
|
|
return asyncio.new_event_loop().run_until_complete(coro)
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def p_patterns_env(isolated_workflows_data, reset_scheduler_state, monkeypatch, tmp_path):
|
|
from backend.apps.agents import agent_manager as p_am
|
|
from backend.apps.patterns import store as p_store
|
|
from backend.apps.settings import settings as p_settings
|
|
|
|
sessions_dir = tmp_path / "sessions"
|
|
sessions_dir.mkdir()
|
|
monkeypatch.setattr(p_am, "SESSIONS_DIR", str(sessions_dir))
|
|
monkeypatch.setattr(p_store, "PATTERNS_DIR", str(tmp_path / "patterns"))
|
|
monkeypatch.setattr(p_store, "SUGGESTIONS_FILE", str(tmp_path / "patterns" / "suggestions.json"))
|
|
monkeypatch.setattr(p_store, "STATE_FILE", str(tmp_path / "patterns" / "state.json"))
|
|
monkeypatch.setattr(p_settings, "load_settings", lambda: SimpleNamespace(pattern_suggestions_enabled=True))
|
|
yield sessions_dir
|
|
|
|
|
|
def p_write_session(sessions_dir, session_id: str, created_at: datetime, name: str = "Inbox check",
|
|
first_msg: str = "summarize my inbox", **extra) -> str:
|
|
doc = {
|
|
"name": name,
|
|
"created_at": created_at.isoformat(),
|
|
"messages": [{"role": "user", "content": first_msg}] if first_msg else [],
|
|
"browser_domains": extra.pop("browser_domains", []),
|
|
}
|
|
doc.update(extra)
|
|
(sessions_dir / f"{session_id}.json").write_text(json.dumps(doc))
|
|
return session_id
|
|
|
|
|
|
def p_pattern_times(n: int = 4) -> list[datetime]:
|
|
# Same weekday + hour, spread over n weeks, all safely in the past.
|
|
return [datetime.now() - timedelta(days=7 * k, hours=1) for k in range(n)]
|
|
|
|
|
|
def test_gather_excludes_what_must_never_count(p_patterns_env):
|
|
from backend.apps.patterns.miner import gather_evidence
|
|
|
|
now = datetime.now()
|
|
p_write_session(p_patterns_env, "keepme", now - timedelta(days=1))
|
|
p_write_session(p_patterns_env, "subagent", now - timedelta(days=1), parent_session_id="parent1")
|
|
p_write_session(p_patterns_env, "wfrun", now - timedelta(days=1), workflow_run_id="run1")
|
|
p_write_session(p_patterns_env, "stale", now - timedelta(days=45))
|
|
p_write_session(p_patterns_env, "automated", now - timedelta(days=1), name="Morning brief")
|
|
p_write_session(p_patterns_env, "nomsg", now - timedelta(days=1), first_msg="")
|
|
|
|
evidence = gather_evidence(automated_titles=["Morning brief"])
|
|
assert [e.id for e in evidence] == ["keepme"]
|
|
|
|
|
|
def test_cadence_computed_in_code():
|
|
from backend.apps.patterns.miner import compute_cadence
|
|
|
|
weekly_times = p_pattern_times(4)
|
|
cadence = compute_cadence(weekly_times)
|
|
assert cadence.kind == "weekly"
|
|
assert cadence.on_days == [(weekly_times[0].weekday() + 1) % 7]
|
|
assert cadence.hour == weekly_times[0].hour
|
|
|
|
daily_times = [datetime(2026, 7, d, 9, 0) for d in range(10, 16)]
|
|
assert compute_cadence(daily_times).kind == "daily"
|
|
|
|
scattered = [datetime(2026, 7, 6, 9), datetime(2026, 7, 14, 15), datetime(2026, 7, 22, 20)]
|
|
assert compute_cadence(scattered).kind == "irregular"
|
|
|
|
|
|
def test_parse_drops_fabricated_evidence():
|
|
from backend.apps.patterns.miner import SessionEvidence, parse_suggestions
|
|
|
|
by_id = {
|
|
f"s{i}": SessionEvidence(id=f"s{i}", created_at=datetime.now() - timedelta(days=i),
|
|
title="t", first_message="m", domains=[])
|
|
for i in range(4)
|
|
}
|
|
raw = json.dumps([
|
|
{ # only 2 of its claimed ids exist -> dropped despite claiming 4
|
|
"description": "You often do a fabricated thing",
|
|
"session_ids": ["s0", "s1", "ghost1", "ghost2"],
|
|
"workflow_title": "Fabricated thing",
|
|
"workflow_steps": ["do it"],
|
|
},
|
|
{
|
|
"description": "You often summarize your inbox in the morning",
|
|
"session_ids": ["s0", "s1", "s2", "s3"],
|
|
"workflow_title": "Morning inbox summary",
|
|
"workflow_steps": ["Summarize the inbox"],
|
|
},
|
|
])
|
|
out = parse_suggestions(raw, by_id)
|
|
assert len(out) == 1
|
|
assert out[0].evidence_count == 4 # our count from verified ids, not the model's
|
|
assert out[0].workflow_title == "Morning inbox summary"
|
|
|
|
assert parse_suggestions("total garbage", by_id) == []
|
|
assert parse_suggestions("```json\n[]\n```", by_id) == []
|
|
|
|
|
|
def p_seed_pattern(sessions_dir) -> list[str]:
|
|
ids = []
|
|
for i, t in enumerate(p_pattern_times(4)):
|
|
ids.append(p_write_session(sessions_dir, f"pat{i}", t, name="AI news rundown",
|
|
first_msg="give me a rundown of today's AI news"))
|
|
# Filler so MIN_SESSIONS_TO_MINE is met.
|
|
for i in range(8):
|
|
p_write_session(sessions_dir, f"fill{i}", datetime.now() - timedelta(days=i + 1, hours=3),
|
|
name=f"One-off {i}", first_msg=f"random question {i}")
|
|
return ids
|
|
|
|
|
|
def p_miner_json(ids: list[str]) -> str:
|
|
return json.dumps([{
|
|
"description": "You often ask for a rundown of AI news",
|
|
"session_ids": ids,
|
|
"workflow_title": "Daily AI news rundown",
|
|
"workflow_steps": ["Gather today's AI news and summarize the top stories"],
|
|
}])
|
|
|
|
|
|
def test_mining_pass_end_to_end(p_patterns_env, monkeypatch):
|
|
from backend.apps.patterns import miner, store
|
|
|
|
ids = p_seed_pattern(p_patterns_env)
|
|
|
|
async def p_fake_aux(lines, automated_titles, declined):
|
|
return p_miner_json(ids)
|
|
|
|
monkeypatch.setattr(miner, "p_call_miner", p_fake_aux)
|
|
assert p_run(miner.run_mining_pass(force=True)) == 1
|
|
pending = store.pending_suggestions()
|
|
assert len(pending) == 1
|
|
assert pending[0].evidence_count == 4
|
|
assert pending[0].cadence.kind == "weekly"
|
|
|
|
# A near-identical pattern is never offered twice.
|
|
assert p_run(miner.run_mining_pass(force=True)) == 0
|
|
|
|
# The daily throttle blocks an unforced pass outright.
|
|
assert p_run(miner.run_mining_pass(force=False)) == 0
|
|
|
|
|
|
def test_dismissed_signature_never_returns(p_patterns_env, monkeypatch):
|
|
from backend.apps.patterns import miner, store
|
|
|
|
ids = p_seed_pattern(p_patterns_env)
|
|
|
|
async def p_fake_aux(lines, automated_titles, declined):
|
|
return p_miner_json(ids)
|
|
|
|
monkeypatch.setattr(miner, "p_call_miner", p_fake_aux)
|
|
p_run(miner.run_mining_pass(force=True))
|
|
suggestion = store.pending_suggestions()[0]
|
|
suggestion.status = "dismissed"
|
|
store.update_suggestion(suggestion)
|
|
|
|
assert p_run(miner.run_mining_pass(force=True)) == 0
|
|
assert store.pending_suggestions() == []
|
|
|
|
|
|
def test_kill_switch_gates_the_miner_itself(p_patterns_env, monkeypatch):
|
|
from backend.apps.patterns import miner
|
|
from backend.apps.settings import settings as p_settings
|
|
|
|
p_seed_pattern(p_patterns_env)
|
|
monkeypatch.setattr(p_settings, "load_settings", lambda: SimpleNamespace(pattern_suggestions_enabled=False))
|
|
|
|
async def p_fail_if_called(lines, automated_titles, declined):
|
|
raise AssertionError("miner ran despite kill switch")
|
|
|
|
monkeypatch.setattr(miner, "p_call_miner", p_fail_if_called)
|
|
assert p_run(miner.run_mining_pass(force=True)) == 0
|
|
|
|
|
|
def test_accept_creates_real_scheduled_workflow(p_patterns_env, monkeypatch):
|
|
from backend.apps.agents.core.ws_manager import ws_manager
|
|
from backend.apps.patterns import miner, store
|
|
from backend.apps.patterns.patterns import accept_suggestion, dismiss_suggestion
|
|
from backend.apps.workflows import storage as wf_storage
|
|
from backend.apps.workflows import workflows as wf_routes
|
|
|
|
async def p_noop(*args, **kwargs):
|
|
return None
|
|
|
|
monkeypatch.setattr(ws_manager, "broadcast_global", p_noop)
|
|
|
|
async def p_no_meta(wf):
|
|
return "", "", []
|
|
|
|
monkeypatch.setattr(wf_routes, "_generate_workflow_metadata", p_no_meta)
|
|
|
|
ids = p_seed_pattern(p_patterns_env)
|
|
|
|
async def p_fake_aux(lines, automated_titles, declined):
|
|
return p_miner_json(ids)
|
|
|
|
monkeypatch.setattr(miner, "p_call_miner", p_fake_aux)
|
|
p_run(miner.run_mining_pass(force=True))
|
|
suggestion = store.pending_suggestions()[0]
|
|
|
|
result = p_run(accept_suggestion(suggestion.id))
|
|
wf = wf_storage.get_workflow(result["workflow"]["id"])
|
|
assert wf is not None
|
|
assert wf.title == "Daily AI news rundown"
|
|
assert wf.schedule.enabled is True
|
|
assert wf.schedule.repeat_unit == "week"
|
|
assert wf.steps[0].text.startswith("Gather today's AI news")
|
|
assert store.get_suggestion(suggestion.id).status == "accepted"
|
|
assert store.get_suggestion(suggestion.id).workflow_id == wf.id
|
|
|
|
# Accepted or dismissed suggestions can't be acted on twice.
|
|
from fastapi import HTTPException
|
|
with pytest.raises(HTTPException):
|
|
p_run(dismiss_suggestion(suggestion.id))
|