vrr/ouroboros
2
0
Fork 0
mirror of https://github.com/razzant/ouroboros.git synced 2026-10-03 04:07:04 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 7

Pin the alarm clock, the wake message and retarget the retired-loop harnesses

Browse source 
tests/test_consciousness.py is rewritten for tick(now): every branch
(disabled, live wake, live owner turn, not due, allowance unknown/exhausted,
no owner chat, launch with the envelope and rendered message, refusal,
finish -> chosen interval / pending reason / doubling backoff, notify
debounce, the same-instant admission race, self-trigger exclusion, boot floor,
the one-time inbox archive, graceful stop of a live wake, the snapshot keys).
tests/test_consciousness_wake.py pins the renderer (placeholders, event
truncation, fail-soft readers), the template's hints and the server status
projection.

Retired with the loop: test_consciousness_observations.py,
test_observation_fold.py, test_background_model_wait_browser.py, the
background twins in test_openai_chat_dispatch*.py, test_anthropic_native_custody.py,
test_transport_death_retry.py, the background-role pins in
test_owner_live_delivery.py / test_quiz_answer.py, the BG budget pins and the
consciousness max_tokens row. The context harnesses (test_repo_read_limits,
test_context, test_durable_learning_completeness) now assert the same
governance and gap facts on the wake's Main context through build_llm_messages;
test_background_model_wait.py keeps only the generic model-wait facts. The
docs-sync identifier audit resolves the wake template against the full
registry plus its render placeholders; the people/identity prompt pins follow
the new wording; test_settings_honesty pins that the wake-up bounds apply
without a restart; the persistence scan and the smoke tool inventory absorb
the retired fold segments and the registered set_next_wakeup.

Co-authored-by: Ouroboros <311266734+ouroboros-agent@users.noreply.github.com>
This commit is contained in:
Ouroboros 2026-09-16 06:03:21 +03:00
parent 8212eebed3
commit c7cd252f2a
25 changed files with 773 additions and 1531 deletions
Download patch file Download diff file Expand all files Collapse all files

65
tests/test_anthropic_native_custody.py
View file

@ -6,7 +6,6 @@ import copy
import hashlib
import json
import pathlib
from types import SimpleNamespace
import pytest
@ -512,67 +511,3 @@ def test_main_preserves_private_receipt_but_persists_only_public_projection(
assert all("opaque thought" not in json.dumps(item) for item in persisted)
assert all(ANTHROPIC_NATIVE_RECEIPT_KEY not in json.dumps(item) for item in persisted)
def test_background_preserves_receipt_and_aggregates_round_disclosures(
tmp_path, monkeypatch,
):
import ouroboros.consciousness as consciousness
import ouroboros.llm_observability as observed
native = _canonical_message()
seen_messages = []
events = []
responses = [
(native, {
"cost": 0.0,
"cost_final": True,
"request_wire": {"attempt_id": "a", "candidate_sha256": "sha-a"},
}),
({"role": "assistant", "content": "done"}, {
"cost": 0.0,
"cost_final": True,
"request_wire": {"attempt_id": "b", "candidate_sha256": "sha-b"},
}),
]
def chat_observed(_llm, **kwargs):
seen_messages.append(copy.deepcopy(kwargs["messages"]))
return responses[len(seen_messages) - 1]
monkeypatch.setattr(observed, "chat_observed", chat_observed)
monkeypatch.setattr(consciousness, "get_consciousness_model", lambda: "anthropic/claude-future")
monkeypatch.setattr(
consciousness,
"append_jsonl",
lambda _path, row: (events.append(row) or True),
)
monkeypatch.setattr(consciousness.BackgroundConsciousness, "_build_context", lambda _self: "ctx")
monkeypatch.setattr(consciousness.BackgroundConsciousness, "_tool_schemas", lambda _self: [])
monkeypatch.setattr(consciousness.BackgroundConsciousness, "_check_budget", lambda _self: True)
monkeypatch.setattr(consciousness.BackgroundConsciousness, "_emit_live_log", lambda *_a, **_k: None)
monkeypatch.setattr(consciousness.BackgroundConsciousness, "_emit_progress", lambda *_a: None)
monkeypatch.setattr(
consciousness.BackgroundConsciousness,
"_execute_tool",
lambda _self, _call, _events, _validation: "tool-result",
)
bg = object.__new__(consciousness.BackgroundConsciousness)
bg._drive_root = tmp_path
bg._llm = SimpleNamespace(_resolve_remote_target=lambda _model: {
"provider": "anthropic",
"resolved_model": "claude-future",
"usage_model": "anthropic/claude-future",
"base_url": "https://api.anthropic.com",
})
bg._max_bg_rounds = 2
bg._paused = False
bg._event_queue = None
bg._bg_spent_usd = 0.0
bg._last_idle_reason = ""
bg._next_wakeup_sec = 1.0
assert bg._think_scoped() is True
assert ANTHROPIC_NATIVE_RECEIPT_KEY in seen_messages[1][2]
thought = next(row for row in events if row.get("type") == "consciousness_thought")
assert thought["request_wire"]["attempt_id"] == "b"
assert [item["attempt_id"] for item in thought["request_wire_history"]] == ["a", "b"]

338
tests/test_background_model_wait.py
View file

@ -1,195 +1,22 @@
"""Real background loop and model-wait controls over a controlled model engine."""
"""Model-wait facts that survive the Background Consciousness redesign: an owner
without a task ceiling keeps lexical deadlines, an image window is not a cycle
deadline, and an auth wait preserves the profile intent. (A wake-up is an ordinary
Main turn now; its model waits are the direct turn's own — see test_model_wait.)"""
from copy import deepcopy
import json
import queue
import threading
import time
from types import SimpleNamespace
import pytest
from ouroboros import config, model_wait
from ouroboros import llm_claudexor as transport
from ouroboros.consciousness import BackgroundConsciousness
from ouroboros.gateway import task_model_wait as gateway
from tests.test_llm_claudexor import Gateway, MODEL, result, ledger
from tests.test_llm_claudexor import MODEL
def until(predicate, timeout=5):
deadline = time.monotonic() + timeout
while time.monotonic() < deadline:
if predicate():
return
time.sleep(0.005)
raise AssertionError("controlled background state did not arrive")
@pytest.fixture
def background(tmp_path, monkeypatch):
root = tmp_path / "data"
monkeypatch.setenv("OUROBOROS_DATA_DIR", str(root))
monkeypatch.setenv("OUROBOROS_SETTINGS_PATH", str(root / "settings.json"))
monkeypatch.setenv("TOTAL_BUDGET", "100")
monkeypatch.setenv("OUROBOROS_MODEL_CONSCIOUSNESS", MODEL)
monkeypatch.setattr(config, "CLAUDEXOR_MODEL_POLL_INTERVAL_SEC", 0.005)
monkeypatch.setattr(config, "NETWORK_WAIT_BACKOFF_START_SEC", 0.005)
monkeypatch.setattr(config, "NETWORK_WAIT_BACKOFF_MAX_SEC", 0.01)
registry = SimpleNamespace(_ctx=SimpleNamespace(task_metadata={}),
get_timeout=lambda _name: 10, schemas=lambda: [])
tools = []
registry.execute = lambda name, args: tools.append((name, args)) or "read completed once"
from ouroboros.tools.tool_result import ToolResult
registry.execute_result = lambda name, args: (
tools.append((name, args))
or ToolResult(status="ok", code="OK", text="read completed once", meta={"source": "fixture"})
)
monkeypatch.setattr(BackgroundConsciousness, "_build_registry", lambda self: registry)
events = queue.Queue()
bc = BackgroundConsciousness(root, tmp_path / "repo", events, lambda: 1)
monkeypatch.setattr(bc, "_build_context", lambda **kwargs: "Own background context")
monkeypatch.setattr(bc, "_tool_schemas", lambda: [])
engine = Gateway()
first = result()
first["message"] = {"role": "assistant", "content": "read first", "tool_calls": [
{"id": "read-once", "type": "function", "function": {"name": "read_file", "arguments": '{"path":"file"}'}}]}
refusal = result(outcome="failed", problem={"code": "subscription_window_exhausted", "message": "controlled quota"})
final = result()
final["message"] = {"role": "assistant", "content": "Cycle truly finished"}
engine.results = [first, refusal, final]
engine.dispatch = ["response_received", "not_started", "response_received"]
monkeypatch.setattr(transport, "ensure_owned_gateway", lambda: engine)
ready = threading.Event()
metadata_ready = threading.Event()
monkeypatch.setattr(bc._llm, "claudexor_model_sources", lambda: {
"sources": [{"id": "codex", "credentialHarness": "fixture"}]})
def catalog(*_, **_kw):
# Source discovery publishes a second revision before this read. Tests
# need that stable snapshot before exercising the real decision ingress.
metadata_ready.set()
return {"source": "codex", "models": [{"id": "exact-model"}] if ready.is_set() else []}
monkeypatch.setattr(bc._llm, "claudexor_model_catalog", catalog)
bc.inject_observation("one pending observation", observation_id="one")
outcomes, failures = [], []
def run():
try:
outcomes.append(bc._think())
except BaseException as error:
failures.append(error)
def start():
bc._running = True
bc._stop_event.clear()
metadata_ready.clear()
bc._thread = threading.Thread(target=run)
bc._thread.start()
return bc._thread
yield SimpleNamespace(bc=bc, engine=engine, ready=ready, metadata_ready=metadata_ready, events=events, tools=tools,
outcomes=outcomes, failures=failures, root=root, start=start,
decide=lambda body: gateway._decide(root, body, get_background_model_wait=bc.live_model_wait))
bc._stop_event.set()
bc._wakeup_event.set()
if bc._thread:
bc._thread.join(5)
assert not bc._thread.is_alive()
bc._tool_executor.shutdown(wait=True, cancel_futures=True)
def waiting(fixture):
if not fixture.metadata_ready.is_set():
return None
rows = fixture.bc.model_wait_snapshot()["model_waits"]
return next((row for row in rows.values() if row["state"] == "waiting"), None)
def decision(row, action="retry", **fields):
return {"request_id": "owner-choice", "decision_id": f"model_wait:bg-consciousness:{row['wait_id']}",
"revision": row["revision"], "action": action, **fields}
@pytest.mark.parametrize("code", ["subscription_window_exhausted", "auth_required"])
def test_background_quota_holds_exact_cycle_without_task_record(background, code):
f = background
f.engine.results[1]["problem"]["code"] = code
thread = f.start()
until(lambda: waiting(f))
owner = f.bc.live_model_wait()
assert owner.execution_window_remaining() is None
assert waiting(f)["worker_slot_held"] is False
assert waiting(f)["reason"] == ("auth" if code == "auth_required" else "quota")
assert f.tools == [("read_file", {"path": "file"})]
assert f.outcomes == [] and len(f.bc._snapshot_pending_observations()) == 1
f.ready.set()
thread.join(5)
assert f.failures == [] and f.outcomes == [True]
assert len(f.tools) == 1 and len(f.engine.creates) == 3
assert f.bc._snapshot_pending_observations() == []
assert not (f.root / "task_results" / "bg-consciousness.json").exists()
assert owner.closed and f.bc.live_model_wait() is None
def test_foreground_pause_after_quota_preserves_old_call_until_resume(background):
f = background
thread = f.start()
until(lambda: waiting(f))
f.bc.pause()
f.ready.set()
until(lambda: not waiting(f))
assert thread.is_alive() and f.outcomes == []
assert len(f.engine.creates) == 2 and len(f.tools) == 1
f.bc.resume()
thread.join(5)
assert f.failures == [] and f.outcomes == [True]
assert len(f.engine.creates) == 3 and len(f.tools) == 1
def test_stop_during_model_wait_keeps_observations_unacknowledged(background):
f = background
thread = f.start()
until(lambda: waiting(f))
owner = f.bc.live_model_wait()
f.bc.stop()
thread.join(5)
assert f.failures == [] and f.outcomes == [False]
assert f.bc._last_idle_reason == "stopped" and owner.closed
assert len(f.engine.creates) == 2 and len(f.bc._snapshot_pending_observations()) == 1
assert not (f.root / "task_results" / "bg-consciousness.json").exists()
def test_unknown_outcome_is_not_waitable_or_free(background):
f = background
f.engine.results[1] = result(outcome="unknown", problem={"code": "provider_unavailable", "message": "unknown"})
f.engine.dispatch[1] = "unknown"
f.start().join(5)
assert f.failures == [] and f.outcomes == [False]
assert len(f.engine.creates) == 2 and len(f.tools) == 1
assert ledger(f.root)[-1]["state"] == "unresolved"
assert len(f.bc._snapshot_pending_observations()) == 1
def test_stop_of_inflight_background_operation_cancels_same_id_once(background):
f = background
f.engine.pending = True
thread = f.start()
until(lambda: bool(f.engine.reads))
f.bc.stop()
thread.join(5)
assert f.failures == [] and f.outcomes == [False]
assert f.bc._last_idle_reason == "stopped"
assert len(f.engine.creates) == 1 and f.engine.cancels == [("op-0", "host_cancelled")]
assert ledger(f.root)[-1]["state"] == "unresolved"
assert len(f.bc._snapshot_pending_observations()) == 1
def test_image_window_fallback_does_not_create_background_cycle_deadline(tmp_path, monkeypatch):
def test_image_window_fallback_does_not_create_a_ceilingless_owner_deadline(tmp_path, monkeypatch):
from ouroboros.tools.vision import _vision_execution_window
monkeypatch.setattr(config, "get_task_abs_ceiling_sec", lambda: 900)
with model_wait.task_model_wait_scope(task={"id": "background"}, drive_root=tmp_path,
with model_wait.task_model_wait_scope(task={"id": "ceilingless"}, drive_root=tmp_path,
event_queue=None, worker_slot_held=False, owner_control=lambda: None) as owner:
owner.started_monotonic = 0
assert owner.execution_window_remaining() is None
@ -197,13 +24,13 @@ def test_image_window_fallback_does_not_create_background_cycle_deadline(tmp_pat
assert owner.control_reason() is None
def test_background_owner_retains_lexical_deadlines_without_task_ceiling(tmp_path, monkeypatch):
owner = model_wait.TaskModelWait(task={"id": "bg-consciousness"}, drive_root=tmp_path,
def test_ceilingless_owner_retains_lexical_deadlines_without_task_ceiling(tmp_path, monkeypatch):
owner = model_wait.TaskModelWait(task={"id": "ceilingless"}, drive_root=tmp_path,
event_queue=None, worker_slot_held=False, owner_control=lambda: None)
owner.started_monotonic = 0
monkeypatch.setattr(config, "get_task_abs_ceiling_sec", lambda: 0)
assert owner.control_reason() is None and owner.execution_window_remaining() is None
with model_wait.task_model_wait_scope(task={"id": "bg-consciousness"}, drive_root=tmp_path,
with model_wait.task_model_wait_scope(task={"id": "ceilingless"}, drive_root=tmp_path,
event_queue=None, worker_slot_held=False, owner_control=lambda: None) as bound:
with model_wait.execution_deadline_scope(model_wait.monotonic_now() - 1):
assert bound.control_reason() == "execution_deadline"
@ -211,129 +38,6 @@ def test_background_owner_retains_lexical_deadlines_without_task_ceiling(tmp_pat
assert bound.control_reason() == "deadline"
def test_owner_switch_is_same_live_map_and_old_cycle_action_is_refused(background):
f = background
thread = f.start()
until(lambda: waiting(f))
row = deepcopy(waiting(f))
owner = f.bc.live_model_wait()
owned_row = owner.waits[row["wait_id"]]
response = f.decide(decision(row, "switch", model=MODEL, credential_profile_id="replacement",
use_local=False, persist_role=False))
assert response.status_code == 202
assert owner.waits[row["wait_id"]] is owned_row
thread.join(5)
assert f.failures == [] and f.outcomes == [True]
assert f.engine.uploads[-1][0]["account"] == {"mode": "pin", "profileId": "replacement"}
assert f.decide(decision(row)).status_code == 409
assert not (f.root / "settings.json").exists()
def test_persistent_background_switch_saves_only_consciousness(background, monkeypatch):
f = background
path = f.root / "settings.json"
monkeypatch.setattr(config, "SETTINGS_PATH", path)
path.write_text(json.dumps({"OUROBOROS_MODEL": "main-stays", "OUROBOROS_MODEL_LIGHT": "light-stays"}))
thread = f.start()
until(lambda: waiting(f))
response = f.decide(decision(deepcopy(waiting(f)), "switch", model=MODEL,
credential_profile_id="replacement", use_local=False, persist_role=True))
assert response.status_code == 202 and json.loads(response.body)["saved"] is True
thread.join(5)
saved = json.loads(path.read_text())
assert saved["OUROBOROS_MODEL"] == "main-stays" and saved["OUROBOROS_MODEL_LIGHT"] == "light-stays"
assert saved["OUROBOROS_MODEL_CONSCIOUSNESS"] == MODEL
assert json.loads(saved["OUROBOROS_MODEL_ACCOUNTS"])["consciousness"] == "replacement"
assert f.outcomes == [True] and not f.failures
@pytest.mark.parametrize("phase", ["before", "after"])
def test_current_background_owner_fences_settings_after_live_owner_changes(background, monkeypatch, phase):
from ouroboros.gateway import owner_settings
f = background
monkeypatch.setattr(config, "SETTINGS_PATH", f.root / "settings.json")
thread = f.start()
until(lambda: waiting(f))
old_owner = f.bc.live_model_wait()
original = owner_settings._owner_update_settings
def stale_before_write(transform, **kwargs):
if phase == "before":
f.bc._model_wait = None
value = original(transform, **kwargs)
f.bc._model_wait = None
return value
monkeypatch.setattr(owner_settings, "_owner_update_settings", stale_before_write)
response = f.decide(decision(deepcopy(waiting(f)), "switch", model=MODEL,
credential_profile_id="replacement", use_local=False, persist_role=True))
assert response.status_code == 409 and json.loads(response.body)["saved"] is (phase == "after")
assert (f.root / "settings.json").exists() is (phase == "after")
f.bc._model_wait = old_owner
f.bc.stop()
thread.join(5)
def test_background_wait_forwarding_and_reload_use_existing_owner(background):
from ouroboros.gateway.history import _assemble_history_response
from ouroboros.utils import append_jsonl
from supervisor.task_model_wait import handle_task_model_wait
f = background
thread = f.start()
until(lambda: (waiting(f) or {}).get("revision"))
owner = f.bc.live_model_wait()
# Source metadata can publish a newer revision while the cycle waits.
# Keep this snapshot current throughout forwarding and reload assertions.
with owner.lock:
row = deepcopy(waiting(f))
event = {"type": "task_model_wait", "task_id": "bg-consciousness", "ts": "2026-09-07T00:00:00Z", **row}
forwarded = []
ctx = SimpleNamespace(RUNNING={}, DRIVE_ROOT=f.root, consciousness=f.bc,
append_jsonl=append_jsonl, bridge=SimpleNamespace(push_log=forwarded.append))
handle_task_model_wait(event, ctx)
assert len(forwarded) == 1 and forwarded[0]["chat_id"] == 1
handle_task_model_wait({**event, "model_wait_owner_id": "previous-cycle"}, ctx)
assert len(forwarded) == 1 and ctx.RUNNING == {}
append_jsonl(f.root / "logs" / "progress.jsonl", {"task_id": "bg-consciousness",
"type": "send_message", "is_progress": True, "ts": "2026-09-07T00:00:00Z", "text": "Earlier thought"})
payload = json.loads(_assemble_history_response(f.root, 1, 10, 10, owner.snapshot()))
current = payload["messages"][-1]
assert current["model_wait_live"] and current["model_wait_owner_id"] == owner.owner_id
assert not any(row.get("task_terminal_status") for row in payload["messages"] if row.get("is_progress"))
f.bc._owner_chat_id_fn = lambda: 0
handle_task_model_wait(event, ctx)
assert forwarded[-1]["chat_id"] == 0
hidden = json.loads(_assemble_history_response(f.root, 1, 10, 10, {**owner.snapshot(), "chat_id": 0}))
assert not any(row.get("model_wait_live") for row in hidden["messages"])
assert any(row.get("task_terminal_status") == "done" for row in hidden["messages"] if row.get("is_progress"))
f.bc.stop()
thread.join(5)
def test_previous_cycle_decision_cannot_target_a_new_live_cycle(background):
f = background
thread = f.start()
until(lambda: waiting(f))
previous = deepcopy(waiting(f))
previous_owner = f.bc.live_model_wait()
f.ready.set()
thread.join(5)
f.ready.clear()
f.engine.results.extend(deepcopy(f.engine.results))
f.engine.dispatch.extend(list(f.engine.dispatch))
thread = f.start()
until(lambda: waiting(f))
current = f.bc.live_model_wait()
assert current is not previous_owner and current.owner_id != previous_owner.owner_id
assert f.decide(decision(previous)).status_code == 404
assert "pending_action" not in waiting(f)
assert len(f.engine.creates) == 5
f.bc.stop()
thread.join(5)
@pytest.mark.parametrize("override,problem_context,expected", [
(None, {}, "configured-pin"), ("", {}, ""),
(None, {"credentialProfileId": "proved-profile"}, "proved-profile"),
@ -351,25 +55,3 @@ def test_auth_wait_profile_hint_preserves_intent_without_inventing_route(tmp_pat
row, = waiter.waits.values()
assert row["credential_profile_id"] == expected
assert error.route == {} # The UI hint is not claimed as an actual provider route.
@pytest.mark.serial
@pytest.mark.parametrize("surface", ["web", "host"])
def test_real_background_owner_is_shared_by_both_decision_transports(background, surface):
from tests.test_model_wait import _decision_clients
f = background
thread = f.start()
until(lambda: waiting(f))
row = deepcopy(waiting(f))
owner = f.bc.live_model_wait()
with _decision_clients(f.root, f.bc.live_model_wait) as clients:
response = clients[surface](decision(row, "switch", model=MODEL,
credential_profile_id="replacement", use_local=False, persist_role=False))
assert response.status_code == 202 and response.json()["saved"] is False
thread.join(5)
assert f.failures == [] and f.outcomes == [True] and owner.closed
assert f.engine.uploads[-1][0]["account"] == {"mode": "pin", "profileId": "replacement"}
stale = clients[surface](decision(row))
assert stale.status_code == 409 and stale.json()["reason_code"] == "task_not_live"
assert not (f.root / "task_results" / "bg-consciousness.json").exists()

7
tests/test_bugfixes_airi.py
View file

@ -34,13 +34,6 @@ def _read(rel: str) -> str:
# ───────────────── Bug 1 (polish): background card finalizes ─────────────────
def test_consciousness_emits_structured_idle_marker_not_text_matched():
src = _read("ouroboros/consciousness.py")
assert "_emit_cycle_idle" in src
assert "consciousness_state" in src
# The marker is structured, never a regex on log text (BIBLE P5).
assert "Going back to sleep" not in src
def test_log_events_derives_bg_card_phase_from_marker():
src = _read("web/modules/log_events.js")

9
tests/test_chat_inject_attachments.py
View file

@ -202,22 +202,19 @@ def test_decision_route_relays_to_the_one_decision_ingress(tmp_path, monkeypatch
captured = {}
async def fake_answer(drive_root, body, *, get_background_model_wait=None):
async def fake_answer(drive_root, body):
captured["drive_root"] = drive_root
captured["body"] = body
captured["get_background_model_wait"] = get_background_model_wait
return 200, {"ok": True, "decision_id": body["decision_id"], "state": "answered", "answered_index": 1}
monkeypatch.setattr(td, "answer_decision", fake_answer)
client = _client(tmp_path, FakeBridge())
getter = lambda: None
client.app.state.get_background_model_wait = getter
body = {"request_id": "tg:99", "decision_id": "quiz:task-1:q1", "option_index": 1}
response = client.post("/chat/decision", headers={"X-Skill-Token": "token"}, json=body)
assert response.status_code == 200
assert response.json() == {"ok": True, "decision_id": "quiz:task-1:q1", "state": "answered", "answered_index": 1}
assert captured == {"drive_root": tmp_path, "body": body, "get_background_model_wait": getter}
assert captured == {"drive_root": tmp_path, "body": body}
def test_decision_route_returns_the_ingress_refusals_verbatim(tmp_path):
@ -236,7 +233,7 @@ def test_decision_route_returns_the_ingress_refusals_verbatim(tmp_path):
def test_decision_route_is_rate_limited_per_skill(tmp_path, monkeypatch):
import ouroboros.gateway.task_decision as td
async def fake_answer(drive_root, body, *, get_background_model_wait=None):
async def fake_answer(drive_root, body):
return 200, {"ok": True}
monkeypatch.setattr(td, "answer_decision", fake_answer)

636
tests/test_consciousness.py
View file

@ -1,318 +1,378 @@
"""The alarm clock of Background Consciousness (redesign P2): ``tick`` decides WHEN an
ordinary Main turn starts on its own; the direct-activity census is the only liveness
truth; ``notify`` debounces early wakes arithmetically; a wake's own finish never
re-arms it. The lane that runs the turn is pinned in ``test_consciousness_wake_lane.py``.
"""
Tests for BackgroundConsciousness helpers.
Verifies progress events have the correct shape, reach the queue,
and respect pause / chat_id=None semantics. Also covers backlog digest
inclusion in background context.
Run: pytest tests/test_consciousness.py -v
"""
from __future__ import annotations
import json
import os
import pathlib
import queue
import sys
import tempfile
import unittest
from unittest.mock import MagicMock, patch
from types import SimpleNamespace
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..'))
import pytest
from ouroboros import consciousness as clock_module
from ouroboros.consciousness import (
ARCHIVED_INBOX_REL, INTERVAL_STATE_KEY, LEGACY_INBOX_REL, NEXT_WAKE_STATE_KEY, BackgroundConsciousness,
)
from supervisor.active_activity import get_direct_activity_registry
T0 = 1_800_000_000.0
FLOOR, CEILING, DEFAULT = 900, 14400, 3300
AVAILABLE = {"status": "available", "limit_usd": 20.0, "accounted_usd": 2.5, "remaining_usd": 17.5,
"resets_at": "", "unknown_unmetered": 0}
class TestEmitProgress(unittest.TestCase):
"""Tests for BackgroundConsciousness._emit_progress."""
@pytest.fixture
def clock(monkeypatch, tmp_path):
"""A constructed alarm clock over an in-memory runtime state and a stubbed lane."""
from supervisor import state, workers
def _make_consciousness(self, chat_id=42, event_queue=None):
"""Create a BackgroundConsciousness with mocked dependencies."""
from ouroboros.consciousness import BackgroundConsciousness
store = {"bg_consciousness_enabled": True, "owner_chat_id": 7}
monkeypatch.setattr(state, "load_state", lambda: dict(store))
tmpdir = tempfile.mkdtemp()
drive_root = pathlib.Path(tmpdir)
(drive_root / "logs").mkdir(parents=True, exist_ok=True)
repo_dir = pathlib.Path(tmpdir) / "repo"
repo_dir.mkdir()
def update_state(mutator):
mutator(store)
return dict(store)
eq = event_queue if event_queue is not None else queue.Queue()
monkeypatch.setattr(state, "update_state", update_state)
monkeypatch.setenv("OUROBOROS_BG_WAKEUP_MIN", str(FLOOR))
monkeypatch.setenv("OUROBOROS_BG_WAKEUP_MAX", str(CEILING))
monkeypatch.setenv("OUROBOROS_CONSCIOUSNESS_AUTONOMY", "act")
monkeypatch.setenv("OUROBOROS_PER_TASK_COST_USD", "50")
monkeypatch.delenv("OUROBOROS_CONSCIOUSNESS_MAX_TASKS", raising=False)
(tmp_path / "logs").mkdir()
(tmp_path / "repo" / "prompts").mkdir(parents=True)
(tmp_path / "repo" / "prompts" / "CONSCIOUSNESS.md").write_text(
"[Wake-up · {reason}] since {last_wake_ago}: {events} level={level} withheld={withheld_tools} "
"spent={spent_usd}/{daily_usd} running={running}/{max_tasks} interval={interval} line={level_line}",
encoding="utf-8")
launches: list = []
receipt = {"admitted": True, "task_id": "wake0001", "reason": ""}
with patch.object(BackgroundConsciousness, '_build_registry', return_value=MagicMock()):
bc = BackgroundConsciousness(
drive_root=drive_root,
repo_dir=repo_dir,
event_queue=eq,
owner_chat_id_fn=lambda: chat_id,
)
return bc, eq, drive_root
def handle_wake_direct(chat_id, text, task_metadata, on_finished=None):
launches.append({"chat_id": chat_id, "text": text, "metadata": task_metadata, "on_finished": on_finished})
return dict(receipt)
def test_event_shape(self):
"""Event has type, chat_id, text, is_progress, ts."""
bc, eq, _ = self._make_consciousness(chat_id=99)
bc._emit_progress("thinking about things")
evt = eq.get_nowait()
self.assertEqual(evt["type"], "send_message")
self.assertEqual(evt["chat_id"], 99)
self.assertEqual(evt["text"], "💬 thinking about things")
self.assertEqual(evt["format"], "markdown")
self.assertTrue(evt["is_progress"])
self.assertIn("ts", evt)
def test_empty_content_skipped(self):
"""Empty or whitespace-only content produces no event."""
bc, eq, drive_root = self._make_consciousness()
progress_path = drive_root / "logs" / "progress.jsonl"
bc._emit_progress("")
bc._emit_progress(" ")
bc._emit_progress(None)
self.assertTrue(eq.empty())
# Also should not persist to file
self.assertFalse(progress_path.exists())
def test_chat_id_none_skips_queue_but_persists(self):
"""When chat_id is None, event is NOT queued but IS persisted."""
bc, eq, drive_root = self._make_consciousness(chat_id=None)
bc._emit_progress("background thought")
# Queue should be empty
self.assertTrue(eq.empty())
# File should have the entry
progress_path = drive_root / "logs" / "progress.jsonl"
self.assertTrue(progress_path.exists())
entry = json.loads(progress_path.read_text().strip())
self.assertEqual(entry["type"], "send_message")
self.assertEqual(entry["content"], "background thought")
self.assertTrue(entry["is_progress"])
def test_paused_events_go_to_deferred(self):
"""When paused, events go to _deferred_events, not the queue."""
bc, eq, _ = self._make_consciousness()
bc._paused = True
bc._emit_progress("deferred thought")
self.assertTrue(eq.empty())
self.assertEqual(len(bc._deferred_events), 1)
self.assertEqual(bc._deferred_events[0]["type"], "send_message")
self.assertEqual(bc._deferred_events[0]["text"], "💬 deferred thought")
monkeypatch.setattr(workers, "handle_wake_direct", handle_wake_direct)
monkeypatch.setattr(clock_module, "allowance_window", lambda root, now=None: dict(AVAILABLE))
monkeypatch.setattr(BackgroundConsciousness, "_running_roots", staticmethod(lambda: 1))
get_direct_activity_registry().clear()
clock = BackgroundConsciousness(tmp_path, tmp_path / "repo", lambda: store.get("owner_chat_id"), now=T0)
yield SimpleNamespace(clock=clock, store=store, launches=launches, receipt=receipt, root=tmp_path)
get_direct_activity_registry().clear()
class TestBackgroundContext(unittest.TestCase):
def test_build_context_includes_improvement_backlog_digest(self):
from ouroboros.consciousness import BackgroundConsciousness
tmpdir = pathlib.Path(tempfile.mkdtemp())
drive_root = tmpdir / "drive"
repo_dir = tmpdir / "repo"
(repo_dir / "prompts").mkdir(parents=True, exist_ok=True)
(repo_dir / "docs").mkdir(parents=True, exist_ok=True)
(drive_root / "memory" / "knowledge").mkdir(parents=True, exist_ok=True)
(drive_root / "logs").mkdir(parents=True, exist_ok=True)
(drive_root / "state").mkdir(parents=True, exist_ok=True)
(repo_dir / "prompts" / "CONSCIOUSNESS.md").write_text("Consciousness prompt", encoding="utf-8")
(repo_dir / "BIBLE.md").write_text("Bible", encoding="utf-8")
(repo_dir / "VERSION").write_text("1.2.3", encoding="utf-8")
(repo_dir / "pyproject.toml").write_text('version = "1.2.3"', encoding="utf-8")
(repo_dir / "README.md").write_text("README", encoding="utf-8")
(repo_dir / "docs" / "ARCHITECTURE.md").write_text('# Ouroboros v1.2.3', encoding="utf-8")
(repo_dir / "docs" / "DEVELOPMENT.md").write_text('# Dev', encoding="utf-8")
(drive_root / "state" / "state.json").write_text('{"spent_usd": 0}', encoding="utf-8")
(drive_root / "memory" / "identity.md").write_text("I am Ouroboros", encoding="utf-8")
(drive_root / "memory" / "scratchpad.md").write_text("scratchpad", encoding="utf-8")
(drive_root / "memory" / "knowledge" / "improvement-backlog.md").write_text(
"# Improvement Backlog\n\n### ibl-1\n- status: open\n- created_at: 2026-04-14T09:00:00+00:00\n- source: execution_reflection\n- category: process\n- task_id: task-1\n- requires_plan_review: yes\n- fingerprint: fp-1\n- summary: Reduce recurring task friction around REVIEW_BLOCKED\n",
encoding="utf-8",
)
for name in ("chat.jsonl", "progress.jsonl", "tools.jsonl", "events.jsonl", "supervisor.jsonl", "task_reflections.jsonl"):
(drive_root / "logs" / name).write_text("", encoding="utf-8")
with patch.object(BackgroundConsciousness, '_build_registry', return_value=MagicMock()):
bc = BackgroundConsciousness(
drive_root=drive_root,
repo_dir=repo_dir,
event_queue=None,
owner_chat_id_fn=lambda: None,
)
text = bc._build_context()
self.assertIn("## Improvement Backlog", text)
self.assertIn("Reduce recurring task friction around REVIEW_BLOCKED", text)
def _events(root):
path = root / "logs" / "events.jsonl"
return [json.loads(line) for line in path.read_text(encoding="utf-8").splitlines() if line.strip()] if path.exists() else []
class TestBackgroundConsciousnessToolScope(unittest.TestCase):
def test_background_consciousness_cannot_execute_or_delegate(self):
from ouroboros.consciousness import BackgroundConsciousness
tmpdir = pathlib.Path(tempfile.mkdtemp())
drive_root = tmpdir / "drive"
repo_dir = tmpdir / "repo"
(drive_root / "logs").mkdir(parents=True, exist_ok=True)
repo_dir.mkdir(parents=True, exist_ok=True)
eq = queue.Queue()
bc = BackgroundConsciousness(
drive_root=drive_root,
repo_dir=repo_dir,
event_queue=eq,
owner_chat_id_fn=lambda: 42,
)
schema_names = {s.get("function", {}).get("name") for s in bc._tool_schemas()}
self.assertIn("send_user_message", schema_names)
self.assertIn("update_identity", schema_names)
self.assertIn("recent_tasks", schema_names)
self.assertNotIn("schedule_subagent", schema_names)
self.assertNotIn("get_task_result", schema_names)
self.assertNotIn("wait_task", schema_names)
self.assertNotIn("wait_tasks", schema_names)
self.assertNotIn("run_command", schema_names)
self.assertNotIn("commit_reviewed", schema_names)
def test_set_next_wakeup_schema_follows_configured_bounds(self):
"""The advertised range is the LIVE clamp, not a constant: with
OUROBOROS_BG_WAKEUP_MIN/MAX overridden the schema must say so, because
the handler clamps to those values (prompt-audit review finding)."""
import os
from unittest import mock
from ouroboros.consciousness import BackgroundConsciousness
tmpdir = pathlib.Path(tempfile.mkdtemp())
drive_root = tmpdir / "drive"
repo_dir = tmpdir / "repo"
(drive_root / "logs").mkdir(parents=True, exist_ok=True)
repo_dir.mkdir(parents=True, exist_ok=True)
with mock.patch.dict(os.environ, {"OUROBOROS_BG_WAKEUP_MIN": "60", "OUROBOROS_BG_WAKEUP_MAX": "3600"}):
bc = BackgroundConsciousness(
drive_root=drive_root,
repo_dir=repo_dir,
event_queue=queue.Queue(),
owner_chat_id_fn=lambda: 42,
)
schema = next(
s["function"] for s in bc._tool_schemas() if s.get("function", {}).get("name") == "set_next_wakeup"
)
self.assertIn("60-3600", schema["description"])
self.assertIn("60-3600", schema["parameters"]["properties"]["seconds"]["description"])
self.assertEqual(bc._wakeup_min, 60)
self.assertEqual(bc._wakeup_max, 3600)
def _register(task_id, *, initiator=""):
actor = SimpleNamespace(_busy=True, _current_task_id=task_id, _accepting_owner_messages=True,
_current_task_metadata={"initiator": initiator} if initiator else {},
_current_chat_id=7, _current_task_text="x", _task_started_ts=T0)
return get_direct_activity_registry().register(task_id, 7, actor=actor)
class TestBackgroundConsciousnessCost(unittest.TestCase):
def test_think_accepts_unlimited_total_budget(self):
from ouroboros.consciousness import BackgroundConsciousness
from ouroboros.settings_setup_contract import resolve_total_budget_usd
from ouroboros.usage_accounting import current_usage_scope
bc = object.__new__(BackgroundConsciousness)
bc._drive_root = pathlib.Path(tempfile.mkdtemp())
bc._bg_budget_pct = 5.0
captured = []
bc._think_scoped = lambda: captured.append(current_usage_scope()) or True
with patch.dict(os.environ, {"TOTAL_BUDGET": "0"}):
self.assertIsNone(resolve_total_budget_usd())
self.assertTrue(bc._think())
self.assertIsNone(captured[0].global_limit_usd)
self.assertIsNone(captured[0].root_limit_usd)
def test_unknown_round_cost_stays_nullable_in_durable_thought(self):
from ouroboros.consciousness import BackgroundConsciousness
tmpdir = pathlib.Path(tempfile.mkdtemp())
drive_root = tmpdir / "drive"
repo_dir = tmpdir / "repo"
(drive_root / "logs").mkdir(parents=True)
repo_dir.mkdir()
with patch.object(BackgroundConsciousness, "_build_registry", return_value=MagicMock()):
bc = BackgroundConsciousness(
drive_root=drive_root,
repo_dir=repo_dir,
event_queue=None,
owner_chat_id_fn=lambda: None,
)
with (
patch.object(bc, "_build_context", return_value="context"),
patch.object(bc, "_tool_schemas", return_value=[]),
patch.object(bc, "_check_budget", return_value=True),
patch(
"ouroboros.llm_observability.chat_observed",
return_value=({"content": "thought"}, {"cost": None}),
),
):
self.assertTrue(bc._think_scoped())
events = [
json.loads(line)
for line in (drive_root / "logs" / "events.jsonl").read_text().splitlines()
]
thought = next(event for event in events if event.get("type") == "consciousness_thought")
self.assertIsNone(thought["cost_usd"])
self.assertFalse(thought["cost_final"])
# --- boot -----------------------------------------------------------------------
def test_boot_floor_never_wakes_in_the_first_second(clock):
"""An overdue persisted next_wake_at is pushed past now + floor; a later one is kept."""
assert clock.clock.next_wake_at == T0 + FLOOR
clock.store[NEXT_WAKE_STATE_KEY] = T0 + 5000
later = BackgroundConsciousness(clock.root, clock.root / "repo", lambda: 7, now=T0)
assert later.next_wake_at == T0 + 5000
assert later.enabled is True
class TestWakeScopedTransportState(unittest.TestCase):
"""One Background wake is one cache identity and one active transport turn."""
def _make(self):
from ouroboros.consciousness import BackgroundConsciousness
def test_legacy_inbox_is_archived_once_without_being_read(clock):
inbox = clock.root / LEGACY_INBOX_REL
inbox.parent.mkdir(parents=True, exist_ok=True)
inbox.write_bytes(b'{"op":"enqueue"\n not json at all\n')
BackgroundConsciousness(clock.root, clock.root / "repo", lambda: 7, now=T0)
assert not inbox.exists()
assert (clock.root / ARCHIVED_INBOX_REL).read_bytes() == b'{"op":"enqueue"\n not json at all\n'
# A second legacy file (an older archive already there) gets its own name, nothing is overwritten.
inbox.write_bytes(b"second")
BackgroundConsciousness(clock.root, clock.root / "repo", lambda: 7, now=T0 + 1)
assert (clock.root / ARCHIVED_INBOX_REL).read_bytes().startswith(b'{"op":"enqueue"')
assert (clock.root / "archive" / f"consciousness_observations_{int(T0 + 1)}.jsonl").read_bytes() == b"second"
tmpdir = pathlib.Path(tempfile.mkdtemp())
drive_root, repo_dir = tmpdir / "drive", tmpdir / "repo"
(drive_root / "logs").mkdir(parents=True)
repo_dir.mkdir()
with patch.object(BackgroundConsciousness, "_build_registry", return_value=MagicMock()):
bc = BackgroundConsciousness(drive_root=drive_root, repo_dir=repo_dir,
event_queue=None, owner_chat_id_fn=lambda: None)
bc._running = True
return bc
def test_each_wake_owns_one_slot_and_key_and_clears_both(self):
bc = self._make()
seen = []
# --- tick branches ----------------------------------------------------------------
def scoped():
seen.append((bc._model_turn_state, bc._wake_cache_affinity, bc._model_wait.owner_id))
return True
with patch.object(bc, "_think_scoped", side_effect=scoped):
self.assertTrue(bc._think())
self.assertTrue(bc._think())
first, second = seen
# The cache identity IS the wake's existing model-wait owner id.
self.assertEqual(first[1], first[2])
self.assertEqual(second[1], second[2])
self.assertIsNone(first[0].envelope)
self.assertIsNot(first[0], second[0])
self.assertNotEqual(first[1], second[1])
self.assertIsNone(bc._model_turn_state)
self.assertEqual(bc._wake_cache_affinity, "")
def test_disabled_clock_does_nothing(clock):
clock.clock.stop()
assert clock.clock.tick(T0 + FLOOR + 1) == "disabled"
assert clock.launches == []
def test_a_wake_round_sends_its_own_key_and_slot(self):
bc = self._make()
sent = []
def observed(_llm, **kwargs):
sent.append((kwargs, bc._wake_cache_affinity, bc._model_turn_state, bc._model_wait.owner_id))
return {"content": "thought"}, {"cost": None}
def test_not_due_before_next_wake_at(clock):
assert clock.clock.tick(T0 + FLOOR - 1) == "not_due"
assert clock.launches == []
with (
patch.object(bc, "_build_context", return_value="context"),
patch.object(bc, "_tool_schemas", return_value=[]),
patch.object(bc, "_check_budget", return_value=True),
patch("ouroboros.llm_observability.chat_observed", side_effect=observed),
):
self.assertTrue(bc._think())
self.assertEqual(len(sent), 1)
kwargs, key, slot, owner_id = sent[0]
self.assertEqual(kwargs["cache_affinity"], owner_id)
self.assertEqual(key, owner_id)
self.assertIs(kwargs["model_turn_state"], slot)
if __name__ == "__main__":
unittest.main()
def test_a_live_wake_or_owner_turn_defers_the_wake(clock):
_register("owner001")
assert clock.clock.tick(T0 + FLOOR + 1) == "owner_turn_live"
get_direct_activity_registry().clear()
_register("wakeXYZ", initiator="consciousness")
assert clock.clock.tick(T0 + FLOOR + 1) == "wake_live"
assert clock.clock.status_snapshot()["live_wake_task_id"] == "wakeXYZ"
assert clock.launches == []
def test_allowance_unknown_skips_with_a_typed_status_and_the_floor(clock, monkeypatch):
monkeypatch.setattr(clock_module, "allowance_window",
lambda root, now=None: {"status": "allowance_unknown", "error": "OSError: ledger"})
now = T0 + FLOOR + 1
assert clock.clock.tick(now) == "skipped:allowance_unknown"
assert clock.clock.next_wake_at == now + FLOOR
assert clock.store[NEXT_WAKE_STATE_KEY] == now + FLOOR
snapshot = clock.clock.status_snapshot()
assert snapshot["last_wake_outcome"] == "skipped:allowance_unknown" and "OSError" in snapshot["last_error"]
assert [row["reason"] for row in _events(clock.root) if row["type"] == "consciousness_wake_skipped"] == ["allowance_unknown"]
assert clock.launches == []
def test_allowance_exhausted_skips_until_the_window_frees(clock, monkeypatch):
from ouroboros.deadline_utils import parse_deadline_ts
resets_at = "2027-02-01T00:00:00+00:00"
monkeypatch.setattr(clock_module, "allowance_window", lambda root, now=None: {
**AVAILABLE, "status": "exhausted", "accounted_usd": 21.0, "remaining_usd": 0.0, "resets_at": resets_at})
assert clock.clock.tick(T0 + FLOOR + 1) == "skipped:allowance_exhausted"
assert clock.clock.next_wake_at == parse_deadline_ts(resets_at).timestamp()
assert clock.clock.status_snapshot()["last_wake_outcome"] == "skipped:allowance_exhausted"
# A reset instant already in the past (or none: DAILY_USD=0) still waits at least the floor.
monkeypatch.setattr(clock_module, "allowance_window", lambda root, now=None: {
**AVAILABLE, "status": "exhausted", "resets_at": ""})
clock.clock._next_wake_at = T0
assert clock.clock.tick(T0 + 5) == "skipped:allowance_exhausted"
assert clock.clock.next_wake_at == T0 + 5 + FLOOR
def test_no_owner_chat_waits_for_the_first_conversation(clock):
clock.store["owner_chat_id"] = None
assert clock.clock.tick(T0 + FLOOR + 1) == "skipped:waiting_for_first_conversation"
assert clock.clock.next_wake_at == T0 + FLOOR + 1 + FLOOR
assert clock.launches == []
# --- launch -----------------------------------------------------------------------
def test_launch_starts_an_ordinary_main_turn_with_the_wake_envelope(clock):
clock.store[INTERVAL_STATE_KEY] = 1200
now = T0 + FLOOR + 1
assert clock.clock.tick(now) == "launched"
[launch] = clock.launches
assert launch["chat_id"] == 7
meta = launch["metadata"]
assert meta["initiator"] == "consciousness" and meta["usage_category"] == "consciousness"
assert meta["wake_reason"] == "heartbeat" and meta["consciousness_autonomy"] == "act"
assert meta["model_role"] == "consciousness" and meta["runtime_mode_cap"] == "light"
assert "toggle_evolution" in meta["disabled_tools"] and "steer_task" not in meta["disabled_tools"]
assert meta["root_cost_ceiling_usd"] == 17.5 # min(per-task cap 50, remaining 17.5)
text = launch["text"]
assert text.startswith("[Wake-up · heartbeat]") and "level=act" in text and "spent=2.50/20.00" in text
assert "running=1/2" in text and "interval=1200" in text and "toggle_evolution" in text
assert "no wake since this process started" in text
assert launch["on_finished"] == clock.clock._wake_finished
snapshot = clock.clock.status_snapshot()
assert snapshot["last_wake_task_id"] == "wake0001" and snapshot["last_wake_outcome"] == "running"
started = [row for row in _events(clock.root) if row["type"] == "consciousness_wake_started"]
assert started and started[0]["task_id"] == "wake0001" and started[0]["wake_reason"] == "heartbeat"
def test_launch_ceiling_is_the_remaining_allowance_when_no_per_task_cap(clock, monkeypatch):
monkeypatch.setenv("OUROBOROS_PER_TASK_COST_USD", "0")
assert clock.clock.tick(T0 + FLOOR + 1) == "launched"
assert clock.launches[0]["metadata"]["root_cost_ceiling_usd"] == 17.5
def test_pending_reason_is_captured_and_cleared_at_launch(clock):
clock.clock.notify("task_finished:abc:completed")
assert clock.clock.tick(T0 + FLOOR + 1) == "launched"
assert clock.launches[0]["metadata"]["wake_reason"] == "task_finished:abc:completed"
assert clock.launches[0]["text"].startswith("[Wake-up · task_finished:abc:completed]")
assert clock.clock.pending_reason is None
def test_rejected_wake_is_typed_and_retried_after_the_floor(clock):
clock.receipt.update({"admitted": False, "task_id": "", "reason": "budget_exhausted"})
now = T0 + FLOOR + 1
assert clock.clock.tick(now) == "rejected:budget_exhausted"
assert clock.clock.next_wake_at == now + FLOOR
assert clock.clock.status_snapshot()["last_wake_outcome"] == "rejected:budget_exhausted"
assert [row["reason"] for row in _events(clock.root) if row["type"] == "consciousness_wake_rejected"] == ["budget_exhausted"]
def test_a_turn_admitted_in_the_same_instant_keeps_the_reason_for_later(clock, monkeypatch):
from supervisor import workers
def handle_wake_direct(*_a, **_k):
raise AssertionError("must not launch beside a live turn")
monkeypatch.setattr(workers, "handle_wake_direct", handle_wake_direct)
calls = {"n": 0}
real = clock.clock.live_turns
def live_turns():
calls["n"] += 1
if calls["n"] == 2: # the re-check under the gate lock sees a just-admitted owner turn
_register("owner002")
return real()
monkeypatch.setattr(clock.clock, "live_turns", live_turns)
clock.clock.notify("project_digest:p1")
assert clock.clock.tick(T0 + FLOOR + 1) == "owner_turn_live"
assert clock.clock.pending_reason == "project_digest:p1"
# --- finish -----------------------------------------------------------------------
def _launched(clock, now=T0 + FLOOR + 1):
assert clock.clock.tick(now) == "launched"
return clock.launches[-1]["on_finished"]
def test_finish_schedules_the_chosen_interval_clamped(clock, monkeypatch):
finished = _launched(clock)
monkeypatch.setattr(clock_module.time, "time", lambda: T0 + 5000)
clock.store[INTERVAL_STATE_KEY] = 100 # below the floor
finished("wake0001", True)
assert clock.clock.next_wake_at == T0 + 5000 + FLOOR
snapshot = clock.clock.status_snapshot()
assert snapshot["last_wake_outcome"] == "done" and snapshot["last_error"] == ""
assert snapshot["last_wake_at"].startswith("2027-")
clock.store[INTERVAL_STATE_KEY] = 10 ** 6 # above the ceiling
finished("wake0001", True)
assert clock.clock.next_wake_at == T0 + 5000 + CEILING
del clock.store[INTERVAL_STATE_KEY] # no choice → the default
finished("wake0001", True)
assert clock.clock.next_wake_at == T0 + 5000 + DEFAULT
def test_finish_with_a_pending_reason_wakes_after_the_floor(clock, monkeypatch):
finished = _launched(clock)
monkeypatch.setattr(clock_module.time, "time", lambda: T0 + 5000)
clock.clock.notify("task_finished:xyz:failed")
finished("wake0001", True)
assert clock.clock.next_wake_at == T0 + 5000 + FLOOR
assert clock.clock.pending_reason == "task_finished:xyz:failed"
def test_runner_failure_backs_off_by_doubling_until_a_wake_succeeds(clock, monkeypatch):
finished = _launched(clock)
monkeypatch.setattr(clock_module.time, "time", lambda: T0 + 5000)
finished("wake0001", False)
assert clock.clock.next_wake_at == T0 + 5000 + DEFAULT * 2
snapshot = clock.clock.status_snapshot()
assert snapshot["last_wake_outcome"] == "failed" and "wake0001" in snapshot["last_error"]
finished("wake0001", False)
assert clock.clock.next_wake_at == T0 + 5000 + DEFAULT * 4
finished("wake0001", False)
assert clock.clock.next_wake_at == T0 + 5000 + CEILING # 3300 * 8 > 14400
finished("wake0001", True)
assert clock.clock.next_wake_at == T0 + 5000 + DEFAULT
# --- notify -----------------------------------------------------------------------
def test_notify_pulls_the_next_wake_to_the_floor_after_the_last_wake(clock, monkeypatch):
clock.clock._next_wake_at = T0 + 3000
monkeypatch.setattr(clock_module.time, "time", lambda: T0 + 100)
clock.clock.notify("task_finished:a:completed")
assert clock.clock.next_wake_at == T0 + 100 # no wake yet: now
clock.clock._last_wake_at = T0
clock.clock._next_wake_at = T0 + 3000
clock.clock.notify("task_finished:b:completed")
assert clock.clock.next_wake_at == T0 + FLOOR # arithmetic debounce off the last wake
clock.clock.notify("task_finished:c:completed")
assert clock.clock.next_wake_at == T0 + FLOOR and clock.clock.pending_reason == "task_finished:c:completed"
# A notify never pushes a nearer wake further away.
clock.clock._next_wake_at = T0 + 50
clock.clock.notify("task_finished:d:completed")
assert clock.clock.next_wake_at == T0 + 50
def test_task_done_notifies_for_roots_of_any_outcome_but_never_for_consciousness_origin():
from supervisor.events_task_done import _notify_consciousness_of_root_done
reasons: list = []
ctx = SimpleNamespace(consciousness=SimpleNamespace(notify=reasons.append))
_notify_consciousness_of_root_done(ctx, {}, None, {"status": "failed"}, {"task_id": "t1", "status": "failed"})
_notify_consciousness_of_root_done(ctx, {"delegation_role": "root"}, None, {}, {"task_id": "t2", "status": "completed"})
_notify_consciousness_of_root_done(ctx, {"delegation_role": "subagent"}, None, {}, {"task_id": "t3", "status": "completed"})
_notify_consciousness_of_root_done(ctx, {}, None, {"metadata": {"initiator": "consciousness"}},
{"task_id": "wake1", "status": "completed"})
_notify_consciousness_of_root_done(ctx, {}, {"initiator": "consciousness", "usage_category": "consciousness_task"}, {},
{"task_id": "started1", "status": "completed"})
assert reasons == ["task_finished:t1:failed", "task_finished:t2:completed"]
# No alarm clock on the ctx (supervisor init failed) is not an error.
_notify_consciousness_of_root_done(SimpleNamespace(), {}, None, {}, {"task_id": "t4", "status": "completed"})
def test_project_digest_and_orphan_heal_reach_notify(monkeypatch, tmp_path):
from ouroboros import server_maintenance
from supervisor.events_project_routing import _handle_project_digest
reasons: list = []
ctx = SimpleNamespace(DRIVE_ROOT=tmp_path, consciousness=SimpleNamespace(notify=reasons.append))
monkeypatch.setattr("ouroboros.projects_registry.touch_project", lambda root, pid: None)
_handle_project_digest({"project_id": "p9", "task_id": "t9"}, ctx)
assert reasons == ["project_digest:p9"]
monkeypatch.setattr("ouroboros.skill_review_runner.reconcile_stale_review_jobs", lambda root: None)
monkeypatch.setattr("ouroboros.task_status.reconcile_orphaned_running_tasks", lambda root, **kw: 2)
monkeypatch.setattr("ouroboros.projects_registry.reconcile_projects", lambda root: None)
monkeypatch.setattr(server_maintenance, "_resume_interrupted_project_deletions", lambda: None)
healed: list = []
server_maintenance._periodic_zombie_reconcile(on_orphans_healed=healed.append)
assert healed == [2]
# --- owner controls ---------------------------------------------------------------
def test_start_and_stop_flip_the_flag_and_stop_arms_a_graceful_stop_of_a_live_wake(clock, monkeypatch):
import threading
assert clock.clock.start() == "Background consciousness is already enabled."
assert clock.clock.stop() == "Background consciousness disabled."
assert clock.clock.enabled is False and clock.clock.tick(T0 + FLOOR + 1) == "disabled"
assert clock.clock.stop() == "Background consciousness is already disabled."
assert clock.clock.start().startswith("Background consciousness enabled; next wake-up at ")
assert clock.clock.enabled is True
_launched(clock)
_register("wake0001", initiator="consciousness")
stopped: list = []
done = threading.Event()
def stop_direct_chat_turn(task_id, turn, **_kw):
stopped.append((task_id, turn["id"]))
done.set()
return "ended"
monkeypatch.setattr("supervisor.worker_chat_lane.stop_direct_chat_turn", stop_direct_chat_turn)
message = clock.clock.stop()
assert message == "Background consciousness disabled; wake-up wake0001 ends at its next step."
assert done.wait(5) and stopped == [("wake0001", "wake0001")]
def test_status_snapshot_carries_the_alarm_facts(clock):
snapshot = clock.clock.status_snapshot()
assert set(snapshot) == {
"enabled", "level", "next_wake_at", "pending_reason", "last_wake_at", "last_wake_task_id",
"last_wake_outcome", "last_error", "spent_24h_usd", "daily_usd", "allowance_resets_at",
"tasks_running", "max_tasks", "live_wake_task_id",
}
assert snapshot["enabled"] is True and snapshot["level"] == "act"
assert snapshot["next_wake_at"].startswith("2027-") and snapshot["last_wake_at"] == ""
assert snapshot["spent_24h_usd"] == 2.5 and snapshot["daily_usd"] == 20.0
assert snapshot["tasks_running"] == 1 and snapshot["max_tasks"] == 2 and snapshot["live_wake_task_id"] == ""

168
tests/test_consciousness_wake.py Normal file
View file

@ -0,0 +1,168 @@
"""The wake-up message and envelope (Background Consciousness redesign P2), and the
server's projection of the alarm clock's snapshot into one honest status line."""
from __future__ import annotations
import json
import pathlib
from types import SimpleNamespace
import pytest
from ouroboros import consciousness_wake as wake
REPO = pathlib.Path(__file__).resolve().parents[1]
T0 = 1_800_000_000.0
def _iso(ts):
return wake._iso(ts)
def _result(root, task_id, *, status="completed", ts, cost=1.25, direct=False, quizzes=None, description=""):
(root / "task_results").mkdir(exist_ok=True)
row = {"task_id": task_id, "status": status, "ts": _iso(ts), "updated_at": _iso(ts), "_schema_version": 1,
"accounted_upper_bound_usd": cost, "description": description or f"do {task_id}",
"metadata": {}, "_is_direct_chat": direct}
if quizzes:
row["owner_quiz"] = quizzes
(root / "task_results" / f"{task_id}.json").write_text(json.dumps(row), encoding="utf-8")
def test_wake_task_metadata_carries_origin_level_and_soft_ceiling(monkeypatch):
monkeypatch.setenv("OUROBOROS_CONSCIOUSNESS_AUTONOMY", "act")
meta = wake.wake_task_metadata("observe", "heartbeat", root_cost_ceiling_usd=3.5)
assert meta["initiator"] == "consciousness" and meta["usage_category"] == "consciousness"
assert meta["consciousness_autonomy"] == "observe" and meta["runtime_mode_cap"] == "light"
assert meta["model_role"] == "consciousness" and meta["wake_reason"] == "heartbeat"
assert "promote_chat_to_task" in meta["disabled_tools"] and meta["root_cost_ceiling_usd"] == 3.5
full = wake.wake_task_metadata("full", "event:x")
assert full["disabled_tools"] == [] and full["runtime_mode_cap"] == "" and "root_cost_ceiling_usd" not in full
assert wake.wake_task_metadata("bogus", "")["consciousness_autonomy"] == "act" # falls back to the setting
def test_events_list_settled_tasks_open_cards_and_owner_messages_since_the_last_wake(tmp_path):
since = T0 - 3600
_result(tmp_path, "old01", ts=since - 10)
_result(tmp_path, "new01", ts=since + 10, status="failed", cost=0.5, description="build the thing")
_result(tmp_path, "run01", ts=since + 20, status="running")
_result(tmp_path, "chat1", ts=since + 30, direct=True) # an owner's own turn: already in Recent chat
_result(tmp_path, "prev1", ts=since + 40) # the previous wake itself is excluded by id
_result(tmp_path, "ask01", ts=since - 100, status="running",
quizzes={"q1": {"state": "open"}, "q2": {"state": "answered"}})
(tmp_path / "logs").mkdir()
(tmp_path / "logs" / "chat.jsonl").write_text("\n".join([
json.dumps({"direction": "in", "ts": _iso(since + 5), "text": "hi"}),
json.dumps({"direction": "out", "ts": _iso(since + 6), "text": "hello"}),
json.dumps({"direction": "in", "ts": _iso(since - 5), "text": "earlier"}),
json.dumps({"direction": "in", "ts": _iso(since + 7), "text": "again"}),
]) + "\n", encoding="utf-8")
lines = wake.wake_events(tmp_path, since=since, now=T0, exclude_task_id="prev1")
assert "- open question card q1 on task ask01 (no answer yet)" in lines
assert "- task new01 failed, $0.50: build the thing" in lines
assert "- 2 message(s) from your human (see Recent chat)" in lines
assert not [line for line in lines if "old01" in line or "run01" in line or "chat1" in line or "prev1" in line]
assert wake.wake_events(tmp_path / "missing", since=since, now=T0) == []
def test_render_substitutes_every_placeholder_and_truncates_events_honestly(tmp_path, monkeypatch):
repo = tmp_path / "repo"
(repo / "prompts").mkdir(parents=True)
(repo / "prompts" / "CONSCIOUSNESS.md").write_text(
(REPO / "prompts" / "CONSCIOUSNESS.md").read_text(encoding="utf-8"), encoding="utf-8")
for index in range(15):
_result(tmp_path, f"t{index:02d}", ts=T0 - 100 + index)
text = wake.render_wake_message(
tmp_path, repo, reason="task_finished:t14:completed", last_wake_at=T0 - 5400, since=T0 - 5400, now=T0,
level="act", disabled_tools=["toggle_evolution", "request_restart"], spent_usd=4.0, daily_usd=20.0,
running=1, max_tasks=2, interval=3300)
for key in wake.PLACEHOLDERS:
assert "{" + key + "}" not in text, key
assert text.startswith("[Wake-up · task_finished:t14:completed]")
assert "1 h 30 min ago" in text and "autonomy act — everything your runtime mode allows except" in text
assert "toggle_evolution, request_restart (calling them is refused)" in text
assert "Allowance (last 24 h): 4.00 / 20.00 USD" in text and "still running: 1/2" in text
assert "wake-up interval is 3300 s" in text
assert text.count("- task t") == wake.EVENT_LINES_MAX and "(+5 more; see recent_tasks)" in text
quiet = wake.render_wake_message(
tmp_path, repo, reason="heartbeat", last_wake_at=0.0, since=T0 + 1, now=T0, level="full",
disabled_tools=[], spent_usd=None, daily_usd=0, running=0, max_tasks=0, interval=900)
assert "no wake since this process started" in quiet and "nothing new" in quiet
assert "withheld at this level: none" in quiet and "Allowance (last 24 h): unknown / 0.00 USD" in quiet
assert "including evolution" in quiet
def test_render_survives_a_missing_template_and_an_unreadable_task_store(tmp_path):
(tmp_path / "task_results").mkdir()
(tmp_path / "task_results" / "broken.json").write_text("{not json", encoding="utf-8")
text = wake.render_wake_message(
tmp_path, tmp_path / "no-repo", reason="heartbeat", last_wake_at=0.0, since=T0 - 1, now=T0, level="act",
disabled_tools=[], spent_usd=0.0, daily_usd=20.0, running=0, max_tasks=2, interval=3300)
assert text.startswith("[Wake-up · heartbeat]") and "{" not in text
def test_template_names_only_its_placeholders_and_the_wake_hints():
template = (REPO / "prompts" / "CONSCIOUSNESS.md").read_text(encoding="utf-8")
import re
assert set(re.findall(r"\{([a-z_]+)\}", template)) == set(wake.PLACEHOLDERS)
for hint in ("Doing nothing is a fine outcome", "ask only when the answer changes what you do",
"say what you assume meanwhile", "choose how long", "do not request an acceptance review",
"One maintenance item per wake is a good rhythm", "`set_next_wakeup`", "Allowance (last 24 h)",
"be brief, no essays unless something matters", "people you talk with"):
assert hint in template, hint
assert "up to 10 rounds" not in template and "300 seconds" not in template
# --- the server projection --------------------------------------------------------
@pytest.fixture
def describe(monkeypatch):
import server
holder = {}
monkeypatch.setattr(server, "_consciousness", SimpleNamespace(status_snapshot=lambda: dict(holder)))
return lambda snapshot, enabled=True: (holder.clear(), holder.update(snapshot), server._describe_bg_consciousness_state(enabled))[2]
BASE = {"enabled": True, "level": "act", "next_wake_at": "2027-01-15T12:30:00+00:00", "pending_reason": "",
"last_wake_at": "", "last_wake_task_id": "", "last_wake_outcome": "", "last_error": "",
"spent_24h_usd": 1.0, "daily_usd": 20.0, "allowance_resets_at": "", "tasks_running": 0, "max_tasks": 2,
"live_wake_task_id": ""}
def test_projection_names_every_honest_status(describe):
import server
assert describe(BASE, enabled=False)["status"] == "disabled"
sleeping = describe(BASE)
assert sleeping["status"] == "sleeping" and sleeping["detail"] == f"Sleeping until {server._clock_of(BASE['next_wake_at'])}."
assert sleeping["next_wake_at"] == BASE["next_wake_at"] and sleeping["enabled"] is True
pending = describe({**BASE, "pending_reason": "task_finished:a:completed"})
assert pending["detail"].endswith(" Early wake pending: task_finished:a:completed.")
thinking = describe({**BASE, "live_wake_task_id": "wake0001"})
assert thinking["status"] == "thinking" and "wake0001" in thinking["detail"]
first = describe({**BASE, "last_wake_outcome": "skipped:waiting_for_first_conversation"})
assert first["status"] == "waiting_for_first_conversation"
exhausted = describe({**BASE, "last_wake_outcome": "skipped:allowance_exhausted", "spent_24h_usd": 21.5,
"allowance_resets_at": "2027-01-15T18:00:00+00:00"})
assert exhausted["status"] == "allowance_exhausted"
assert "$21.50 of $20.00" in exhausted["detail"] and server._clock_of("2027-01-15T18:00:00+00:00") in exhausted["detail"]
unknown = describe({**BASE, "last_wake_outcome": "skipped:allowance_unknown", "last_error": "OSError: ledger"})
assert unknown["status"] == "allowance_unknown" and "OSError: ledger" in unknown["detail"]
rejected = describe({**BASE, "last_wake_outcome": "rejected:budget_exhausted"})
assert rejected["status"] == "wake_rejected" and "budget_exhausted" in rejected["detail"]
failed = describe({**BASE, "last_wake_outcome": "failed", "last_error": "wake-up w1 failed in its runner"})
assert failed["status"] == "wake_failed" and "backing off" in failed["detail"]
done = describe({**BASE, "last_wake_outcome": "done", "last_wake_at": "2027-01-15T11:35:00+00:00"})
assert done["status"] == "sleeping"
def test_projection_without_a_constructed_clock_is_stopped_not_running(monkeypatch):
import server
monkeypatch.setattr(server, "_consciousness", None)
described = server._describe_bg_consciousness_state(True)
assert described["status"] == "stopped" and "not constructed" in described["detail"]
assert server._describe_bg_consciousness_state(False)["status"] == "disabled"

38
tests/test_context.py
View file

@ -492,8 +492,10 @@ def test_health_invariants_come_first_in_dynamic_context(tmp_path):
assert dynamic_text.index("## Health Invariants") < dynamic_text.index("## Drive state")
def test_health_invariants_come_first_in_background_consciousness_context(tmp_path):
from ouroboros.consciousness import BackgroundConsciousness
def test_health_invariants_come_first_in_a_consciousness_wake_context(tmp_path):
"""A wake-up is an ordinary Main turn: the same builder, the same section order."""
from ouroboros.context import build_llm_messages
from ouroboros.memory import Memory
repo_dir = tmp_path / "repo"
drive_root = tmp_path / "drive"
@ -503,7 +505,7 @@ def test_health_invariants_come_first_in_background_consciousness_context(tmp_pa
(drive_root / "logs").mkdir(parents=True, exist_ok=True)
(drive_root / "state").mkdir(parents=True, exist_ok=True)
(repo_dir / "prompts" / "CONSCIOUSNESS.md").write_text("Consciousness prompt", encoding="utf-8")
(repo_dir / "prompts" / "SYSTEM.md").write_text("System prompt", encoding="utf-8")
(repo_dir / "BIBLE.md").write_text("Bible", encoding="utf-8")
(repo_dir / "VERSION").write_text("1.2.3", encoding="utf-8")
(repo_dir / "pyproject.toml").write_text('version = "1.2.3"', encoding="utf-8")
@ -523,15 +525,31 @@ def test_health_invariants_come_first_in_background_consciousness_context(tmp_pa
(drive_root / "logs" / "supervisor.jsonl").write_text("", encoding="utf-8")
(drive_root / "logs" / "task_reflections.jsonl").write_text("", encoding="utf-8")
bg = BackgroundConsciousness(
drive_root=drive_root,
repo_dir=repo_dir,
event_queue=None,
owner_chat_id_fn=lambda: None,
class FakeEnv:
def drive_path(self, p):
return drive_root / p
def repo_path(self, p):
return repo_dir / p
@property
def repo_dir(self):
return repo_dir
@property
def drive_root(self):
return drive_root
messages, _cap_info = build_llm_messages(
env=FakeEnv(),
memory=Memory(drive_root=drive_root, repo_dir=repo_dir),
task={"id": "wake1", "type": "task", "text": "[Wake-up · heartbeat]", "_is_direct_chat": True,
"metadata": {"initiator": "consciousness", "usage_category": "consciousness"}},
)
text = bg._build_context()
assert text.index("## Health Invariants") < text.index("## Drive state")
dynamic_text = messages[0]["content"][2]["text"]
assert dynamic_text.startswith("## Health Invariants")
assert dynamic_text.index("## Health Invariants") < dynamic_text.index("## Drive state")
def test_project_recent_chat_filters_archives_before_recent_bound(tmp_path, monkeypatch):

14
tests/test_context_budget_ssot.py
View file

@ -18,9 +18,8 @@ def _src(rel: str) -> str:
def test_agent_context_budget_values_pinned():
"""Values are the SSOT; changing them is a deliberate, visible edit."""
assert cb.OWNER_LOW_TARGET_TOKENS == 200_000
assert cb.BG_CONTEXT_WARN_CHARS == 600_000
assert cb.BG_CONTEXT_MAX_CHARS == 1_200_000
assert cb.BG_STATE_JSON_WARN_CHARS == 200_000
for retired in ("BG_CONTEXT_WARN_CHARS", "BG_CONTEXT_MAX_CHARS", "BG_STATE_JSON_WARN_CHARS", "BG_OBSERVATIONS_WARN_BYTES"):
assert not hasattr(cb, retired), retired # a wake-up is a Main turn under Main's budgets
assert cb.LARGE_CONTEXT_SECTION_CHARS == 200_000
assert cb.MAX_RECENT_CHAT_TAIL == 1000
assert cb.CHAT_ARCHIVE_SCAN_WARN_BYTES == 100_000_000
@ -83,10 +82,6 @@ def test_call_sites_import_the_ssot_names():
assert "read_unconsolidated_chat" in ctx_recent_src
assert "last_consolidated_offset" in _src("ouroboros/memory.py")
consc_src = _src("ouroboros/consciousness.py")
for name in ("BG_CONTEXT_MAX_CHARS", "BG_CONTEXT_WARN_CHARS", "BG_STATE_JSON_WARN_CHARS"):
assert name in consc_src, f"consciousness.py must consume {name}"
ctx_src = _src("ouroboros/context.py")
assert "LARGE_CONTEXT_SECTION_CHARS" in ctx_src
assert "CONTEXT_SOFT_CAP_TOKENS" not in ctx_src
@ -99,11 +94,6 @@ def test_old_bare_literals_are_gone_from_call_sites():
"""The decisive anti-drift check: no bare literal can outlive the SSOT."""
assert "> 1_200_000" not in _src("ouroboros/loop.py")
consc = _src("ouroboros/consciousness.py")
assert "= 1_200_000" not in consc
assert "= 600_000" not in consc
assert "> 200_000" not in consc
ctx = _src("ouroboros/context.py")
assert "= 200_000" not in ctx
assert "_soft_cap = 200_000" not in _src("ouroboros/agent.py")

52
tests/test_docs_sync.py
View file

@ -361,14 +361,17 @@ def test_chat_id_addressing_docs_match_the_code_that_routes_it():
assert "tests/test_chat_id_truthiness_guard.py" in development
def test_consciousness_prompt_matches_scope_limited_contracts():
def test_consciousness_prompt_is_the_wake_message_of_an_ordinary_main_turn():
"""The wake-up runs on Main's system prompt and tools (owner decision В15); this file
is its USER message: no private capability catalog, no round or interval limits."""
consciousness = _read("prompts/CONSCIOUSNESS.md")
assert "schedule subagents" in consciousness
assert "wait on subagents" in consciousness
assert "Update your scratchpad or identity" in consciousness
assert "Write to my human proactively" in consciousness
assert consciousness.startswith("[Wake-up · {reason}]")
assert "Doing nothing is a fine outcome" in consciousness
assert "`set_next_wakeup`" in consciousness and "`escalate`" in consciousness
assert "recent_tasks" in consciousness
for retired in ("You can:", "up to 10 rounds", "Default wakeup", "background consciousness mode"):
assert retired not in consciousness, retired
def test_phase3_governance_language_is_pinned_without_new_qa_surface():
@ -577,35 +580,23 @@ def _prompt_bare_identifiers(text: str) -> set:
def test_prompt_tool_names_resolve_to_registered_tools(tmp_path):
"""Every backticked snake_case identifier in the three runtime prompts is
either a registered tool (public schema), a background-consciousness tool,
or a documented non-tool identifier. Completeness is deliberately NOT
required (the schemas are the catalog); this only forbids phantoms and
stale spellings, the drift class the prompt audit found in every prompt."""
from ouroboros.consciousness import BackgroundConsciousness
either a registered tool (public schema) or a documented non-tool identifier
(for the wake-up template also one of its own render placeholders).
Completeness is deliberately NOT required (the schemas are the catalog);
this only forbids phantoms and stale spellings, the drift class the prompt
audit found in every prompt. The wake-up runs on the full registry, so its
universe is Main's."""
from ouroboros.consciousness_wake import PLACEHOLDERS
root = pathlib.Path(__file__).resolve().parent.parent
registry = ToolRegistry(repo_dir=tmp_path / "repo", drive_root=tmp_path / "data")
registered = {schema["function"]["name"] for schema in registry.schemas()}
# The background whitelist is not taken on faith: every name in it must be a
# registered public tool or a ToolEntry the consciousness module registers
# itself (set_next_wakeup and friends), otherwise the whitelist has rotted.
consciousness_src = (root / "ouroboros" / "consciousness.py").read_text(encoding="utf-8")
bg_private = set(re.findall(r'ToolEntry\("([a-z0-9_]+)"', consciousness_src))
stale_whitelist = set(BackgroundConsciousness._BG_TOOL_WHITELIST) - registered - bg_private
assert not stale_whitelist, f"_BG_TOOL_WHITELIST names unregistered tools: {sorted(stale_whitelist)}"
universe = (
registered
| set(BackgroundConsciousness._BG_TOOL_WHITELIST)
| PROMPT_NON_TOOL_IDENTIFIERS
)
# CONSCIOUSNESS.md runs on the background registry, which admits ONLY the
# whitelist (consciousness.py _tool_schemas/_execute_tool), so a public tool
# that is not whitelisted is a phantom there.
bg_universe = set(BackgroundConsciousness._BG_TOOL_WHITELIST) | PROMPT_NON_TOOL_IDENTIFIERS
universe = registered | PROMPT_NON_TOOL_IDENTIFIERS
wake_universe = universe | set(PLACEHOLDERS)
for rel, allowed in (
("prompts/SYSTEM.md", universe),
("prompts/SAFETY.md", universe),
("prompts/CONSCIOUSNESS.md", bg_universe),
("prompts/CONSCIOUSNESS.md", wake_universe),
):
text = (root / rel).read_text(encoding="utf-8")
unresolved = _prompt_backticked_identifiers(text) - allowed
@ -613,11 +604,10 @@ def test_prompt_tool_names_resolve_to_registered_tools(tmp_path):
f"{rel} names identifiers that are neither registered tools nor "
f"classified non-tool identifiers: {sorted(unresolved)}"
)
# CONSCIOUSNESS.md writes tool names without backticks; its bare snake_case
# tokens must resolve the same way (the runtime drift check in
# context_health only catches names with known prefixes).
# The wake template also names tools without backticks; its bare snake_case
# tokens (the render placeholders aside) must resolve the same way.
bare = _prompt_bare_identifiers((root / "prompts" / "CONSCIOUSNESS.md").read_text(encoding="utf-8"))
unresolved_bare = bare - bg_universe
unresolved_bare = bare - wake_universe
assert not unresolved_bare, (
f"prompts/CONSCIOUSNESS.md names bare identifiers that are neither registered tools "
f"nor classified non-tool identifiers: {sorted(unresolved_bare)}"

379
tests/test_durable_learning_completeness.py
View file

@ -375,28 +375,43 @@ def test_closed_objective_unavailable_abstains_before_chooser(tmp_path, monkeypa
assert called == []
def _bg_fixture(tmp_path, *, backlog_count=10):
from ouroboros.consciousness import BackgroundConsciousness
from ouroboros.improvement_backlog import append_backlog_items
def _wake_context(tmp_path, *, chat_rows=None):
"""The system text a consciousness wake-up gets: Main's own builder (build_llm_messages)
over a wake-shaped task and the real repository prompts — the typed gap facts the
context builders disclose (recent-chat, dialogue-history and schedule digests) are
what the wake reads; nothing consciousness-specific is layered on top."""
from ouroboros.context import build_llm_messages
from ouroboros.memory import Memory
repo_dir = pathlib.Path(__file__).parents[1]
(tmp_path / "logs").mkdir(parents=True)
(tmp_path / "logs" / "chat.jsonl").write_text(
json.dumps({"chat_id": 1, "direction": "in", "text": "complete recent chat"}) + "\n",
encoding="utf-8",
)
(tmp_path / "state").mkdir(parents=True)
(tmp_path / "state" / "state.json").write_text("{}", encoding="utf-8")
for idx in range(backlog_count):
append_backlog_items(tmp_path, [{
"id": f"ibl-bg-{idx}", "fingerprint": f"fp-bg-{idx}",
"summary": f"background item {idx}", "category": "identity", "source": "reflection",
}])
return BackgroundConsciousness(tmp_path, repo_dir, None, lambda: None)
class FakeEnv:
def drive_path(self, p):
return tmp_path / p
def _tool_call(name, args, call_id):
return {"id": call_id, "function": {"name": name, "arguments": json.dumps(args)}}
def repo_path(self, p):
return repo_dir / p
@property
def repo_dir(self):
return repo_dir
@property
def drive_root(self):
return tmp_path
(tmp_path / "logs").mkdir(parents=True, exist_ok=True)
(tmp_path / "state").mkdir(parents=True, exist_ok=True)
(tmp_path / "memory").mkdir(parents=True, exist_ok=True)
if not (tmp_path / "state" / "state.json").exists():
(tmp_path / "state" / "state.json").write_text("{}", encoding="utf-8")
rows = chat_rows if chat_rows is not None else [json.dumps({"chat_id": 1, "direction": "in", "text": "complete recent chat"})]
(tmp_path / "logs" / "chat.jsonl").write_text("".join(row + "\n" for row in rows), encoding="utf-8")
task = {"id": "wake1", "type": "task", "text": "[Wake-up · heartbeat]", "_is_direct_chat": True,
"metadata": {"initiator": "consciousness", "usage_category": "consciousness",
"consciousness_autonomy": "act"}}
messages, _cap = build_llm_messages(env=FakeEnv(), memory=Memory(drive_root=tmp_path, repo_dir=repo_dir), task=task)
return "\n\n".join(block["text"] for block in messages[0]["content"])
def _write_schedules(tmp_path, count):
@ -404,311 +419,45 @@ def _write_schedules(tmp_path, count):
"id": f"schedule-{idx}", "name": f"schedule {idx}", "enabled": True,
"trigger": {"type": "cron", "expr": "0 * * * *"},
} for idx in range(count)]
(tmp_path / "state").mkdir(parents=True, exist_ok=True)
(tmp_path / "state" / "scheduled_tasks.json").write_text(
json.dumps({"tasks": tasks}), encoding="utf-8",
)
def test_bgc_direct_identity_update_requires_complete_named_omission(tmp_path):
bc = _bg_fixture(tmp_path)
try:
# The knowledge header is separate from the exact body; both the
# delivered body and revision must preserve a CRLF source.
backlog = tmp_path / "memory" / "knowledge" / "improvement-backlog.md"
backlog.write_bytes(
backlog.read_bytes().replace(b"\r\n", b"\n").replace(b"\n", b"\r\n")
)
context = bc._build_context()
assert "knowledge_read" in context and "improvement-backlog" in context
content = "I remain directly self-authoring after complete source materialization."
blocked = bc._execute_tool(_tool_call("update_identity", {"content": content}, "u1"), [])
assert "IDENTITY_UPDATE_ABSTAINED" in blocked
read = bc._execute_tool(_tool_call("knowledge_read", {"topic": "improvement-backlog"}, "r1"), [])
assert "background item 9" in read
updated = bc._execute_tool(_tool_call("update_identity", {"content": content}, "u2"), [])
assert updated.startswith("OK: identity updated")
journal = tmp_path / "memory" / "identity_journal.jsonl"
assert journal.exists() and content in journal.read_text(encoding="utf-8")
finally:
bc._tool_executor.shutdown(wait=False, cancel_futures=True)
def test_wake_context_discloses_a_malformed_recent_chat_gap(tmp_path):
context = _wake_context(tmp_path, chat_rows=[
json.dumps({"chat_id": 1, "direction": "in", "text": "complete recent chat"}),
'{"direction":"in","text":"broken"',
])
assert "jsonl_malformed" in context
def test_bgc_source_mutation_after_read_cannot_authorize_identity_rewrite(tmp_path, monkeypatch):
bc = _bg_fixture(tmp_path)
try:
bc._build_context()
backlog = tmp_path / "memory" / "knowledge" / "improvement-backlog.md"
changed = backlog.read_bytes() + (
b"\n### ibl-concurrent\n- summary: changed after materialization\n"
)
real_read_bytes = pathlib.Path.read_bytes
target_reads = 0
def mutate_before_snapshot(self, *args, **kwargs):
nonlocal target_reads
if self == backlog:
target_reads += 1
if target_reads == 1:
captured = real_read_bytes(self, *args, **kwargs)
self.write_bytes(changed)
return captured
return real_read_bytes(self, *args, **kwargs)
monkeypatch.setattr(pathlib.Path, "read_bytes", mutate_before_snapshot)
bc._execute_tool(_tool_call("knowledge_read", {"topic": "improvement-backlog"}, "r1"), [])
result = bc._execute_tool(
_tool_call("update_identity", {"content": "must remain blocked"}, "u1"), [],
)
assert target_reads >= 2 # reader capture plus current-source validation
assert "IDENTITY_UPDATE_ABSTAINED" in result
finally:
bc._tool_executor.shutdown(wait=False, cancel_futures=True)
def test_wake_context_carries_a_complete_dialogue_block_without_a_gap(tmp_path):
(tmp_path / "memory").mkdir(parents=True, exist_ok=True)
(tmp_path / "memory" / "dialogue_blocks.json").write_text(json.dumps([{
"ts": "2026-08-21T00:00:00Z", "source": "consolidator",
"content": "Complete consolidated biography block.",
}]), encoding="utf-8")
_write_schedules(tmp_path, 8)
context = _wake_context(tmp_path)
assert "Complete consolidated biography block." in context
dialogue = context.split("## Dialogue History", 1)[1].split("\n## ", 1)[0]
assert "[MEMORY GAP]" not in dialogue
def test_bgc_source_snapshot_cannot_mix_text_and_digest(tmp_path, monkeypatch):
bc = _bg_fixture(tmp_path)
try:
backlog = tmp_path / "memory" / "knowledge" / "improvement-backlog.md"
backlog.write_bytes(
backlog.read_bytes().replace(b"\r\n", b"\n").replace(b"\n", b"\r\n")
)
bc._build_context()
original = backlog.read_bytes()
changed = original + b"\r\n### ibl-concurrent\r\n- summary: changed bytes\r\n"
real_read_bytes = pathlib.Path.read_bytes
target_reads = 0
def oscillate_during_validation(self, *args, **kwargs):
nonlocal target_reads
if self == backlog:
target_reads += 1
if target_reads == 1:
self.write_bytes(changed)
raw = real_read_bytes(self, *args, **kwargs)
self.write_bytes(original)
return raw
return real_read_bytes(self, *args, **kwargs)
monkeypatch.setattr(pathlib.Path, "read_bytes", oscillate_during_validation)
materialized = bc._execute_tool(
_tool_call("knowledge_read", {"topic": "improvement-backlog"}, "r1"), [],
)
assert materialized.endswith(changed.decode("utf-8"))
assert materialized.startswith("[Knowledge source]")
assert "improvement-backlog" not in bc._identity_source_reads
backlog.write_bytes(changed)
result = bc._execute_tool(
_tool_call("update_identity", {"content": "must remain blocked"}, "u1"), [],
)
assert "IDENTITY_UPDATE_ABSTAINED" in result
finally:
bc._tool_executor.shutdown(wait=False, cancel_futures=True)
def test_wake_context_discloses_a_durable_dialogue_gap(tmp_path):
(tmp_path / "memory").mkdir(parents=True, exist_ok=True)
(tmp_path / "memory" / "dialogue_blocks.json").write_text(json.dumps([{
"ts": "2026-08-21T00:00:00Z", "source": "consolidator",
"gap_id": "dialogue-gap-123",
"content": "[MEMORY GAP] A durable biography interval is unavailable.",
}]), encoding="utf-8")
context = _wake_context(tmp_path)
assert "## Dialogue History" in context and "[MEMORY GAP]" in context
def test_bgc_unavailable_named_omission_abstains_without_approval_flow(tmp_path):
bc = _bg_fixture(tmp_path)
try:
bc._build_context()
(tmp_path / "memory" / "knowledge" / "improvement-backlog.md").unlink()
content = "I retain direct authority but abstain when the named source is unavailable."
result = bc._execute_tool(_tool_call("update_identity", {"content": content}, "u1"), [])
assert "IDENTITY_UPDATE_ABSTAINED" in result
assert "approval" not in result.lower()
assert not (tmp_path / "memory" / "identity_journal.jsonl").exists()
finally:
bc._tool_executor.shutdown(wait=False, cancel_futures=True)
def test_bgc_malformed_recent_chat_gap_blocks_direct_identity_update(tmp_path):
bc = _bg_fixture(tmp_path, backlog_count=0)
try:
chat = tmp_path / "logs" / "chat.jsonl"
chat.write_text(
chat.read_text(encoding="utf-8") + '{"direction":"in","text":"broken"\n',
encoding="utf-8",
)
context = bc._build_context()
assert "jsonl_malformed" in context
content = "I must not rewrite identity from a context with a known chat gap."
result = bc._execute_tool(_tool_call("update_identity", {"content": content}, "u1"), [])
assert "IDENTITY_UPDATE_ABSTAINED" in result
assert not (tmp_path / "memory" / "identity_journal.jsonl").exists()
finally:
bc._tool_executor.shutdown(wait=False, cancel_futures=True)
def test_bgc_complete_recent_chat_keeps_direct_identity_update_available(tmp_path):
bc = _bg_fixture(tmp_path, backlog_count=0)
try:
memory_dir = tmp_path / "memory"
memory_dir.mkdir(parents=True, exist_ok=True)
(memory_dir / "dialogue_blocks.json").write_text(json.dumps([{
"ts": "2026-08-21T00:00:00Z", "source": "consolidator",
"content": "Complete consolidated biography block.",
}]), encoding="utf-8")
_write_schedules(tmp_path, 8)
bc._build_context()
content = "I retain direct identity authority with complete ordinary context."
result = bc._execute_tool(_tool_call("update_identity", {"content": content}, "u1"), [])
assert result.startswith("OK: identity updated")
assert content in (tmp_path / "memory" / "identity_journal.jsonl").read_text(encoding="utf-8")
finally:
bc._tool_executor.shutdown(wait=False, cancel_futures=True)
def test_bgc_durable_dialogue_gap_blocks_direct_identity_update(tmp_path):
bc = _bg_fixture(tmp_path, backlog_count=0)
try:
memory_dir = tmp_path / "memory"
memory_dir.mkdir(parents=True, exist_ok=True)
(memory_dir / "dialogue_blocks.json").write_text(json.dumps([{
"ts": "2026-08-21T00:00:00Z", "source": "consolidator",
"gap_id": "dialogue-gap-123",
"content": "[MEMORY GAP] A durable biography interval is unavailable.",
}]), encoding="utf-8")
context = bc._build_context()
assert "## Dialogue History" in context and "[MEMORY GAP]" in context
content = "I must not rewrite identity across a known durable biography gap."
result = bc._execute_tool(_tool_call("update_identity", {"content": content}, "u1"), [])
assert "IDENTITY_UPDATE_ABSTAINED" in result
assert "dialogue-gap-123" in result
assert not (memory_dir / "identity_journal.jsonl").exists()
finally:
bc._tool_executor.shutdown(wait=False, cancel_futures=True)
def test_bgc_scheduled_tasks_omission_blocks_direct_identity_update(tmp_path):
bc = _bg_fixture(tmp_path, backlog_count=0)
try:
_write_schedules(tmp_path, 9)
context = bc._build_context()
assert '"omitted_count": 1' in context
content = "I must not rewrite identity from an incomplete standing-schedule digest."
result = bc._execute_tool(_tool_call("update_identity", {"content": content}, "u1"), [])
assert "IDENTITY_UPDATE_ABSTAINED" in result
assert "scheduled-tasks" in result
assert not (tmp_path / "memory" / "identity_journal.jsonl").exists()
finally:
bc._tool_executor.shutdown(wait=False, cancel_futures=True)
def test_bgc_observation_gap_blocks_direct_identity_update(tmp_path):
bc = _bg_fixture(tmp_path, backlog_count=0)
try:
store = tmp_path / "state" / "consciousness_observations.jsonl"
store.write_text(
'{"op":"enqueue","id":"valid","source":"runtime",'
'"kind":"text","time":"2026-08-21T00:00:00Z",'
'"payload":"known row","ref":null}\n'
'{"op":"enqueue","id":"broken"\n',
encoding="utf-8",
)
context = bc._build_context()
assert "source_complete=False" in context
assert "state/consciousness_observations.jsonl" in context
assert "background-observations" in context
content = "I must not rewrite identity across a known observation-source gap."
result = bc._execute_tool(_tool_call("update_identity", {"content": content}, "u1"), [])
assert "IDENTITY_UPDATE_ABSTAINED" in result
assert "background-observations" in result
assert not (tmp_path / "memory" / "identity_journal.jsonl").exists()
finally:
bc._tool_executor.shutdown(wait=False, cancel_futures=True)
def test_bgc_omitted_observation_rows_block_direct_identity_update(tmp_path):
bc = _bg_fixture(tmp_path, backlog_count=0)
try:
for index in range(11):
assert bc.inject_observation(
f"observation-{index}", observation_id=f"omitted-{index}"
)
context = bc._build_context()
assert "omitted=1" in context
assert "source_complete=False" in context
assert "background-observations" in context
result = bc._execute_tool(
_tool_call(
"update_identity",
{"content": "Do not rewrite identity from omitted observations."},
"u1",
),
[],
)
assert "IDENTITY_UPDATE_ABSTAINED" in result
assert "background-observations" in result
assert not (tmp_path / "memory" / "identity_journal.jsonl").exists()
finally:
bc._tool_executor.shutdown(wait=False, cancel_futures=True)
def test_bgc_payload_truncation_blocks_direct_identity_update(tmp_path):
bc = _bg_fixture(tmp_path, backlog_count=0)
try:
assert bc.inject_observation(
"p" * 2_000, observation_id="payload-truncated"
)
context = bc._build_context()
assert "payload omitted" in context
assert "source_complete=False" in context
assert "background-observations" in context
result = bc._execute_tool(
_tool_call(
"update_identity",
{"content": "Do not rewrite identity from a truncated payload."},
"u1",
),
[],
)
assert "IDENTITY_UPDATE_ABSTAINED" in result
assert "background-observations" in result
assert not (tmp_path / "memory" / "identity_journal.jsonl").exists()
finally:
bc._tool_executor.shutdown(wait=False, cancel_futures=True)
def test_bgc_aggregate_projection_truncation_blocks_direct_identity_update(tmp_path):
bc = _bg_fixture(tmp_path, backlog_count=0)
try:
for index in range(10):
assert bc.inject_observation(
"small",
observation_id=f"aggregate-{index}",
source="s" * 1_400,
)
context = bc._build_context()
assert "projection truncated" in context
assert "source_complete=False" in context
assert "background-observations" in context
result = bc._execute_tool(
_tool_call(
"update_identity",
{"content": "Do not rewrite identity from an aggregate projection."},
"u1",
),
[],
)
assert "IDENTITY_UPDATE_ABSTAINED" in result
assert "background-observations" in result
assert not (tmp_path / "memory" / "identity_journal.jsonl").exists()
finally:
bc._tool_executor.shutdown(wait=False, cancel_futures=True)
def test_bgc_complete_observation_source_keeps_direct_identity_update_available(tmp_path):
bc = _bg_fixture(tmp_path, backlog_count=0)
try:
assert bc.inject_observation(
"complete observation", observation_id="complete-observation"
)
context = bc._build_context()
assert "source_complete=True" in context
assert "complete-observation" in context
content = "I retain direct identity authority with a complete observation source."
result = bc._execute_tool(_tool_call("update_identity", {"content": content}, "u1"), [])
assert result.startswith("OK: identity updated")
assert content in (tmp_path / "memory" / "identity_journal.jsonl").read_text(
encoding="utf-8"
)
finally:
bc._tool_executor.shutdown(wait=False, cancel_futures=True)
def test_wake_context_discloses_an_omitted_schedule_count(tmp_path):
_write_schedules(tmp_path, 9)
context = _wake_context(tmp_path)
assert '"omitted_count": 1' in context

29
tests/test_evolution_status.py
View file

@ -1,7 +1,6 @@
"""Tests for evolution/consciousness status snapshots."""
import json
from unittest.mock import MagicMock, patch
from starlette.applications import Starlette
from starlette.routing import Route
@ -81,24 +80,22 @@ def test_evolution_status_reports_budget_stop_when_disabled_after_run(monkeypatc
assert snapshot["budget_remaining_usd"] == 1.25
def test_consciousness_status_snapshot_exposes_runtime_fields():
def test_consciousness_status_snapshot_exposes_the_alarm_facts(monkeypatch, tmp_path):
from ouroboros import consciousness as clock_module
from ouroboros.consciousness import BackgroundConsciousness
from supervisor import state
with patch.object(BackgroundConsciousness, "_build_registry", return_value=MagicMock()):
consciousness = BackgroundConsciousness(
drive_root=MagicMock(),
repo_dir=MagicMock(),
event_queue=None,
owner_chat_id_fn=lambda: 1,
)
monkeypatch.setattr(state, "load_state", lambda: {"bg_consciousness_enabled": True, "owner_chat_id": 1})
monkeypatch.setattr(clock_module, "allowance_window", lambda root, now=None: {
"status": "available", "limit_usd": 20.0, "accounted_usd": 3.0, "remaining_usd": 17.0, "resets_at": ""})
monkeypatch.setattr(BackgroundConsciousness, "_running_roots", staticmethod(lambda: 0))
clock = BackgroundConsciousness(tmp_path, tmp_path / "repo", lambda: 1, now=1_800_000_000.0)
clock.notify("task_finished:t1:completed")
snapshot = clock.status_snapshot()
consciousness.pause()
consciousness._next_wakeup_sec = 180
snapshot = consciousness.status_snapshot()
assert snapshot["paused"] is True
assert snapshot["next_wakeup_sec"] == 180
assert snapshot["last_idle_reason"] == "paused_by_active_task"
assert snapshot["enabled"] is True and snapshot["pending_reason"] == "task_finished:t1:completed"
assert snapshot["next_wake_at"] and snapshot["last_wake_at"] == "" and snapshot["live_wake_task_id"] == ""
assert snapshot["spent_24h_usd"] == 3.0 and snapshot["daily_usd"] == 20.0
def test_evolution_data_strips_legacy_checkpoint_result_status(tmp_path, monkeypatch):

2
tests/test_identity_wording.py
View file

@ -12,7 +12,7 @@ def test_prompts_do_not_infer_current_human_from_authors():
assert "my human" in system
assert "I do not know their name" in system
assert "README, BIBLE, git history, or author" in system
assert "Messages From My Human" in consciousness
assert "your human" in consciousness and "the user" not in consciousness
assert "I do not yet know my human's name or profile" in memory
assert "Anton" not in system
assert "Razzhigaev" not in system

1
tests/test_max_tokens_constants.py
View file

@ -83,7 +83,6 @@ def test_summary_and_background_token_budgets():
"ouroboros/reflection.py": "max_tokens=16384",
"ouroboros/post_task_synthesis.py": "max_tokens=16384",
"ouroboros/tools/skill_publish.py": "max_tokens=8192",
"ouroboros/consciousness.py": "max_tokens=65536",
}
for path, needle in expectations.items():
src = Path(path).read_text(encoding="utf-8").replace(" ", "")

29
tests/test_memory_people_prompts.py
View file

@ -103,30 +103,27 @@ def test_memory_vocabulary_is_not_dressed_up_as_tool_names():
assert f"`{word}`" not in text, f"{rel} backticks {word}"
def test_background_checklist_maintains_understanding_of_people():
consciousness = _read("prompts/CONSCIOUSNESS.md")
item = " ".join(
consciousness.split("3. **Knowledge base gaps**", 1)[1].split("\n\n", 1)[0].split()
).lower()
def test_wake_template_maintains_understanding_of_people():
"""The wake-up message (an ordinary Main turn's user text) keeps the commitment:
revise the existing note about a person rather than minting a new one (P12)."""
template = " ".join(_read("prompts/CONSCIOUSNESS.md").split()).lower()
assert "person" in item or "people" in item
# Revise the current note rather than minting a new one (P12: read first).
assert "knowledge_read" in item and "knowledge_write" in item
assert "overview" in item
assert "people you talk with" in template
assert "knowledge_read" in template and "knowledge_write" in template
assert "rather than minting a new one" in template
def test_background_hygiene_checks_beliefs_about_people_too():
hygiene = _section(_read("prompts/CONSCIOUSNESS.md"), "Memory Hygiene").lower()
def test_wake_template_resolves_contradictions_about_people_too():
template = " ".join(_read("prompts/CONSCIOUSNESS.md").split()).lower()
assert "knowledge notes" in hygiene
assert "people" in hygiene
assert "contradictions" in template and "about the people you talk with" in template
def test_background_prompt_addresses_its_human_not_a_user():
def test_wake_template_addresses_its_human_not_a_user():
consciousness = _read("prompts/CONSCIOUSNESS.md")
assert "Write to my human proactively" in consciousness
assert "Message the user proactively" not in consciousness
assert "write to your human" in consciousness
assert "the user" not in consciousness
def test_reflection_nominates_learning_about_people_as_well_as_itself():

24
tests/test_native_conversation_activity.py
View file

@ -150,25 +150,30 @@ def test_restart_census_keeps_native_execution_after_owner_boundary():
assert _live_running_task_ids(SimpleNamespace(RUNNING={})) == []
def test_consciousness_remains_paused_until_all_native_work_returns():
def _mind():
"""An alarm clock that reads liveness off the census alone (no state, no lane)."""
from ouroboros.consciousness import BackgroundConsciousness
mind = object.__new__(BackgroundConsciousness)
mind._paused = False
mind._last_wake_task_id = ""
return mind
def test_consciousness_sees_an_owner_turn_live_until_all_native_work_returns():
mind = _mind()
registry = get_direct_activity_registry()
registry.register("first", 1)
registry.register("second", 2)
assert mind.is_paused
assert mind.live_turns() == ("", True)
registry.unregister("first")
assert mind.is_paused
assert mind.live_turns() == ("", True)
registry.unregister("second")
assert not mind.is_paused
assert mind.live_turns() == ("", False)
def test_native_post_task_retains_activity_and_delivers_answer_early(monkeypatch, tmp_path):
"""Actual synthesis dispatch must stay owned after the ordinary final answer."""
from ouroboros import agent_task_pipeline as pipeline, post_task_evolution
from ouroboros.consciousness import BackgroundConsciousness
from ouroboros.gateway.settings import _has_running_agent_tasks, _has_started_agent_tasks
from ouroboros.post_task_checkpoint import post_task_synthesis_in_flight
from ouroboros.server_restart import _live_running_task_ids
@ -195,8 +200,7 @@ def test_native_post_task_retains_activity_and_delivers_answer_early(monkeypatch
monkeypatch.setattr(pipeline, name, lambda *a, **kw: None)
monkeypatch.setattr(post_task_evolution, "maybe_promote", lambda *a, **kw: None)
env = SimpleNamespace(repo_dir=tmp_path / "repo", drive_root=tmp_path)
mind = object.__new__(BackgroundConsciousness)
mind._paused = False
mind = _mind()
class Actor:
def handle_task(self, task):
@ -237,7 +241,7 @@ def test_native_post_task_retains_activity_and_delivers_answer_early(monkeypatch
assert workers.drain_repo_writers(0) == [task_id]
assert _live_running_task_ids(SimpleNamespace(RUNNING={})) == [task_id]
assert _has_running_agent_tasks() and _has_started_agent_tasks()
assert mind.is_paused
assert mind.live_turns() == ("", True)
# Production early delivery ran before synthesis. The terminal
# completion stays buffered until post-task work returns.
early = bus.get_nowait()
@ -256,7 +260,7 @@ def test_native_post_task_retains_activity_and_delivers_answer_early(monkeypatch
assert workers.drain_repo_writers(0) == []
assert _live_running_task_ids(SimpleNamespace(RUNNING={})) == []
assert not _has_running_agent_tasks() and not _has_started_agent_tasks()
assert not mind.is_paused
assert mind.live_turns() == ("", False)
assert load_task_result(tmp_path, task_id)["root_phase_checkpoint"]["post_task_synthesis"] == "completed"
# Retained final and early final use the existing delivery identity;
# the supervisor deduplicates them. task_done reaches the bus last.

126
tests/test_openai_chat_dispatch.py
View file

@ -729,132 +729,6 @@ def test_function_origin_remains_schema_tolerant(tmp_path):
assert tools.calls == [("probe", {"marker": "wrong"})]
def test_background_custom_schema_error_never_reaches_registry():
from ouroboros.consciousness import BackgroundConsciousness
message, receipts = _invalid_custom_exchange()
instance = object.__new__(BackgroundConsciousness)
result = instance._execute_tool(message["tool_calls"][0], [], receipts[0])
assert "TOOL_ARG_ERROR" in result
def test_background_two_round_custom_error_continuation(monkeypatch, tmp_path):
from concurrent.futures import ThreadPoolExecutor
from ouroboros import consciousness
from ouroboros.consciousness import BackgroundConsciousness
read_tool = {
"type": "function",
"function": {
"name": "read_file",
"description": "Read one allowed path.",
"parameters": {
"type": "object",
"properties": {"path": {"const": "allowed"}},
"required": ["path"],
"additionalProperties": False,
},
},
}
invalid, receipts = _invalid_custom_exchange(
read_tool,
'{"path":"wrong"}',
)
observed_messages = []
def fake_chat_observed(_client, **kwargs):
observed_messages.append(copy.deepcopy(kwargs["messages"]))
if len(observed_messages) == 1:
return copy.deepcopy(invalid), {
**_receipt_usage(receipts),
"cost": 0.0,
}
return {"role": "assistant", "content": "corrected"}, {"cost": 0.0}
registry_calls = []
registry = SimpleNamespace(
_ctx=SimpleNamespace(),
get_timeout=lambda _name: 1,
execute=lambda name, args: registry_calls.append((name, args)) or "executed",
)
instance = object.__new__(BackgroundConsciousness)
instance._build_context = lambda: "context"
instance._tool_schemas = lambda: [read_tool]
instance._llm = SimpleNamespace(_resolve_remote_target=lambda _model: _target())
instance._drive_root = tmp_path
instance._max_bg_rounds = 2
instance._paused = False
instance._emit_live_log = lambda *_a, **_k: None
instance._emit_progress = lambda _content: None
instance._bg_spent_usd = 0.0
instance._check_budget = lambda: True
instance._event_queue = None
instance._last_idle_reason = ""
instance._next_wakeup_sec = 300
instance._wakeup_max = 3600
instance._owner_chat_id_fn = lambda: None
instance._registry = registry
instance._tool_executor = ThreadPoolExecutor(max_workers=1)
(tmp_path / "logs").mkdir()
monkeypatch.setattr(
consciousness,
"get_consciousness_model",
lambda: "openai::future-model-without-prefix",
)
monkeypatch.setattr(consciousness, "resolve_effort", lambda _slot: "medium")
monkeypatch.setattr(dispatch, "projected_context_size_bytes", lambda *_a, **_k: 1)
monkeypatch.setattr(
"ouroboros.llm_observability.chat_observed",
fake_chat_observed,
)
try:
assert instance._think_scoped() is True
finally:
instance._tool_executor.shutdown(wait=True)
assert registry_calls == []
assert len(observed_messages) == 2
assert observed_messages[1][-1]["role"] == "tool"
assert "TOOL_ARG_ERROR" in observed_messages[1][-1]["content"]
def test_background_admission_counts_physical_custom_projection(monkeypatch, tmp_path):
from ouroboros import consciousness
from ouroboros.consciousness import BackgroundConsciousness
logs = tmp_path / "logs"
logs.mkdir()
instance = object.__new__(BackgroundConsciousness)
instance._build_context = lambda: "context"
instance._tool_schemas = _tools
instance._llm = SimpleNamespace(_resolve_remote_target=lambda _model: _target())
instance._drive_root = tmp_path
instance._max_bg_rounds = 1
instance._paused = False
instance._last_idle_reason = ""
observed = {}
def oversized(messages, tools, **kwargs):
observed.update(kwargs)
assert messages[0]["content"] == "context"
assert tools == _tools()
return consciousness.BG_CONTEXT_MAX_CHARS + 1
monkeypatch.setattr(dispatch, "projected_context_size_bytes", oversized)
monkeypatch.setattr(consciousness, "resolve_effort", lambda _slot: "medium")
monkeypatch.setattr(
consciousness,
"get_consciousness_model",
lambda: "openai::future-model-without-prefix",
)
assert instance._think_scoped() is False
assert instance._last_idle_reason == "context_overflow"
assert observed == {"provider": "openai", "reasoning_effort": "medium"}
def test_structured_compaction_returns_one_bounded_tool_error_continuation(monkeypatch, tmp_path):
message, receipts = _invalid_custom_exchange(
context_compaction._CONTEXT_SUMMARIES_TOOL,

103
tests/test_openai_chat_dispatch_review.py
View file

@ -5,9 +5,7 @@ from __future__ import annotations
import copy
import json
import pathlib
from types import SimpleNamespace
import ouroboros.consciousness as consciousness
import ouroboros.context_compaction as context_compaction
import ouroboros.llm as llm_module
import ouroboros.llm_observability as llm_observability
@ -103,28 +101,6 @@ class _QueuedLLM:
return copy.deepcopy(message), usage
def _background_instance(tmp_path, llm, tools, *, rounds=2):
instance = object.__new__(consciousness.BackgroundConsciousness)
instance._build_context = lambda: "context"
instance._tool_schemas = lambda: copy.deepcopy(tools)
instance._llm = llm
instance._drive_root = tmp_path
instance._max_bg_rounds = rounds
instance._paused = False
instance._emit_live_log = lambda *_a, **_k: None
instance._emit_progress = lambda _content: None
instance._bg_spent_usd = 0.0
instance._check_budget = lambda: True
instance._event_queue = None
instance._last_idle_reason = ""
instance._next_wakeup_sec = 300
instance._wakeup_max = 3600
instance._owner_chat_id_fn = lambda: None
instance._registry = SimpleNamespace(_ctx=SimpleNamespace())
(tmp_path / "logs").mkdir(exist_ok=True)
return instance
def _capture_persistence(monkeypatch):
persisted = []
monkeypatch.setattr(
@ -143,42 +119,6 @@ def _response_payloads(persisted):
]
def _pin_remote_background(monkeypatch):
monkeypatch.setattr(
consciousness,
"get_consciousness_model",
lambda: "openai::future-model-without-prefix",
)
monkeypatch.setattr(consciousness, "resolve_effort", lambda _slot: "medium")
def test_background_chat_observed_persists_public_usage_and_returns_receipt(
monkeypatch,
tmp_path,
):
tool = _read_tool()
invalid, first_usage = _custom_exchange(tool, '{"path":"wrong"}')
llm = _QueuedLLM([
(invalid, first_usage),
({"role": "assistant", "content": "corrected"}, {"cost": 0.0}),
])
instance = _background_instance(tmp_path, llm, [tool])
persisted = _capture_persistence(monkeypatch)
_pin_remote_background(monkeypatch)
assert instance._think_scoped() is True
assert len(llm.requests) == 2
assert llm.requests[1]["messages"][-1]["role"] == "tool"
assert "TOOL_ARG_ERROR" in llm.requests[1]["messages"][-1]["content"]
assert dispatch.CUSTOM_RECEIPTS_USAGE_KEY not in first_usage
response_payloads = _response_payloads(persisted)
assert response_payloads[0]["usage"]["request_wire"]
assert all(
dispatch.CUSTOM_RECEIPTS_USAGE_KEY not in item["usage"]
for item in response_payloads
)
def test_compaction_chat_observed_persists_public_usage_and_returns_receipt(
monkeypatch,
tmp_path,
@ -302,49 +242,6 @@ def test_history_body_error_cannot_advance_the_dialect_ladder():
) is None
def test_background_remeasures_growth_before_every_physical_send(monkeypatch, tmp_path):
tool = _read_tool()
first = {
"role": "assistant",
"content": "",
"tool_calls": [{
"id": "call-large",
"type": "function",
"function": {"name": "read_file", "arguments": '{"path":"allowed"}'},
}],
}
llm = _QueuedLLM([
(first, {"cost": 0.0}),
({"role": "assistant", "content": "must not send"}, {"cost": 0.0}),
])
instance = _background_instance(tmp_path, llm, [tool])
instance._execute_tool = (
lambda *_a, **_k: "x" * (consciousness.BG_CONTEXT_MAX_CHARS + 100)
)
_pin_remote_background(monkeypatch)
measured = []
real_measure = dispatch.projected_context_size_bytes
def measure(*args, **kwargs):
result = real_measure(*args, **kwargs)
measured.append(result)
return result
monkeypatch.setattr(dispatch, "projected_context_size_bytes", measure)
assert instance._think_scoped() is False
assert len(llm.requests) == 1
assert len(measured) == 2
assert measured[0] < consciousness.BG_CONTEXT_MAX_CHARS
assert measured[1] > consciousness.BG_CONTEXT_MAX_CHARS
assert instance._last_idle_reason == "context_overflow"
events = [
json.loads(line)
for line in (tmp_path / "logs" / "events.jsonl").read_text().splitlines()
]
assert events[-1]["type"] == "consciousness_context_overflow"
def _serialized_projection(messages, tools):
return json.dumps(
{"messages": messages, "tools": tools},

22
tests/test_owner_live_delivery.py
View file

@ -48,28 +48,6 @@ class TestDeliverOwnerEvent:
assert mode == "deferred"
assert ctx.pending_events[0]["root_task_id"] == "t-root"
def test_background_consciousness_always_deferred_and_unstamped(self):
from ouroboros.tool_capabilities import BACKGROUND_DELEGATION_ROLE
q = _Queue()
ctx = _ctx(event_queue=q, meta={"delegation_role": BACKGROUND_DELEGATION_ROLE})
mode = deliver_owner_event(ctx, {"type": "send_message", "chat_id": 1, "text": "x"})
assert mode == "deferred"
assert q.items == []
# BG frames stay exactly as before the seam: buffered, no pseudo-lineage.
assert "task_id" not in ctx.pending_events[0]
def test_consciousness_stamps_the_shared_background_role(self):
# Literal-drift pin: the producer (consciousness) and the gate
# (owner_delivery) must share ONE constant, not two literals.
import inspect
from ouroboros import consciousness
src = inspect.getsource(consciousness)
assert "BACKGROUND_DELEGATION_ROLE" in src
assert '"delegation_role": "background"' not in src
def test_retry_duplicates_are_accepted_policy(self):
# A live-delivered frame from attempt 1 is not recalled; a retried
# task re-narrates with a fresh ctx and delivers again. The seam

6
tests/test_persistence_inventory.py
View file

@ -558,7 +558,11 @@ def scan_data_paths(root: pathlib.Path = REPO) -> frozenset[str]:
# 290 -> 291: the supervisor's off-lock projection of live direct-chat roots
# (``state/direct_roots.json``, ``supervisor/direct_roots.py``) is the one new
# durable plane of the structural-health train; it has its own row in section 2.
EXPECTED_SCAN_PATHS = 291 # Linked project knowledge history is a dynamic validated shelf.
# 291 -> 290: the Background Consciousness redesign retired the observation inbox
# and its startup fold; the archive segments ``archive/consciousness_observations_<ts>``
# left the population and the one-time ``archive/consciousness_observations.jsonl``
# move target (``ouroboros/consciousness.py``) took their place.
EXPECTED_SCAN_PATHS = 290 # Linked project knowledge history is a dynamic validated shelf.
# Scanned paths that must always be present — guards the scanner itself
# against a silent regression that would shrink coverage while keeping counts

20
tests/test_quiz_answer.py
View file

@ -423,12 +423,6 @@ def test_escalate_settled_parent_is_a_typed_dead_end(tmp_path, monkeypatch):
assert out.startswith("⚠️ ESCALATE_PARENT_SETTLED")
def test_escalate_background_refused(tmp_path):
ctx = _tool_ctx(tmp_path, task_id="bg-consciousness", role="background")
out = _escalate(ctx, question="?", options=["a", "b"], assumption="a")
assert out.startswith("⚠️ ESCALATE_UNAVAILABLE")
def test_escalate_invalid_payload_is_typed(tmp_path):
ctx = _tool_ctx(tmp_path)
out = _escalate(ctx, question="?", options=["only-one"], assumption="a")
@ -898,14 +892,12 @@ def test_recommended_option_rides_the_card_the_projection_and_the_parent_frame(t
def test_escalate_refusals_are_typed_per_branch_and_a_headless_root_still_asks(tmp_path):
"""Verification only: the three real refusal branches as the predicate is written.
Background consciousness is refused; a live direct conversation (including one with
no continuation owner) is refused; REQUIRED waiting without a live continuation owner
is refused. A headless root without owner_wait_callback is NOT refused for an optional
question: it mints the ordinary card and continues under its assumption."""
background = _tool_ctx(tmp_path, task_id="bg", role="background")
out = _escalate(background, question="?", options=["a", "b"], assumption="a")
assert out.startswith("⚠️ ESCALATE_UNAVAILABLE: background consciousness cannot escalate")
"""Verification only: the real refusal branches as the predicate is written.
A live direct conversation (including one with no continuation owner) is refused;
REQUIRED waiting without a live continuation owner is refused. A headless root
without owner_wait_callback is NOT refused for an optional question: it mints the
ordinary card and continues under its assumption. (A consciousness wake-up is an
ordinary root here — nothing refuses it by role.)"""
direct = _tool_ctx(tmp_path)
direct.is_direct_chat = True
out = _escalate(direct, question="?", options=["a", "b"], assumption="a")

162
tests/test_repo_read_limits.py
View file

@ -4,7 +4,7 @@ Also covers the core governance artifact invariants introduced in the
'Core Governance Artifacts' section of docs/DEVELOPMENT.md:
- repo_read default max_lines raised to 2000 so ARCHITECTURE.md is
readable in a single call.
- BackgroundConsciousness._build_context() includes ARCHITECTURE.md.
- the wake-up's Main context (build_llm_messages) includes ARCHITECTURE.md.
- Triad review prompt includes ARCHITECTURE.md even when not touched.
- DEVELOPMENT.md contains the core governance artifact invariant rule.
"""
@ -228,92 +228,75 @@ def test_repo_read_can_read_architecture_md_in_one_call(tmp_path):
assert f"line {n_lines}" in result
def test_consciousness_context_includes_architecture_md(tmp_path):
"""BackgroundConsciousness._build_context must include ARCHITECTURE.md section."""
import queue
from unittest.mock import patch, MagicMock
def _wake_context(tmp_path):
"""The system text a consciousness wake-up gets: Main's own builder over a wake-shaped task."""
from ouroboros.context import build_llm_messages
from ouroboros.memory import Memory
# Minimal fake repo tree
class FakeEnv:
def drive_path(self, p):
return tmp_path / "data" / p
def repo_path(self, p):
return tmp_path / "repo" / p
@property
def repo_dir(self):
return tmp_path / "repo"
@property
def drive_root(self):
return tmp_path / "data"
drive_root = tmp_path / "data"
for rel in ("logs", "state", "memory"):
(drive_root / rel).mkdir(parents=True, exist_ok=True)
(drive_root / "state" / "state.json").write_text("{}", encoding="utf-8")
(tmp_path / "repo" / "prompts").mkdir(parents=True, exist_ok=True)
(tmp_path / "repo" / "prompts" / "SYSTEM.md").write_text("System prompt", encoding="utf-8")
task = {"id": "wake1", "type": "task", "text": "[Wake-up · heartbeat]", "_is_direct_chat": True,
"metadata": {"initiator": "consciousness", "usage_category": "consciousness",
"consciousness_autonomy": "act"}}
messages, _cap = build_llm_messages(env=FakeEnv(), memory=Memory(drive_root=drive_root, repo_dir=tmp_path / "repo"), task=task)
return "\n\n".join(block["text"] for block in messages[0]["content"])
def test_wake_context_includes_architecture_md(tmp_path):
"""A consciousness wake-up reads the same governance artifacts as any Main turn."""
repo_dir = tmp_path / "repo"
(repo_dir / "docs").mkdir(parents=True)
(repo_dir / "BIBLE.md").write_text("# BIBLE", encoding="utf-8")
(repo_dir / "docs" / "ARCHITECTURE.md").write_text(
"# ARCHITECTURE\n\nThis is the architecture doc.", encoding="utf-8"
)
(repo_dir / "prompts" / "CONSCIOUSNESS.md").mkdir(parents=True, exist_ok=True) if False else None
# Consciousness prompt (optional — can be missing)
drive_root = tmp_path / "data"
(drive_root / "logs").mkdir(parents=True)
(drive_root / "state").mkdir(parents=True)
(drive_root / "state" / "state.json").write_text("{}", encoding="utf-8")
(drive_root / "memory").mkdir(parents=True)
eq = queue.Queue()
from ouroboros.consciousness import BackgroundConsciousness
with patch.object(BackgroundConsciousness, "_build_registry", return_value=MagicMock()), \
patch("ouroboros.consciousness.build_memory_sections", return_value=[]), \
patch("ouroboros.consciousness.build_health_invariants", return_value=""), \
patch("ouroboros.consciousness.build_runtime_section", return_value="## Runtime\n\nok"), \
patch("ouroboros.consciousness.build_recent_sections", return_value=[]):
bc = BackgroundConsciousness(
drive_root=drive_root,
repo_dir=repo_dir,
event_queue=eq,
owner_chat_id_fn=lambda: 1,
)
context = bc._build_context()
context = _wake_context(tmp_path)
assert "## ARCHITECTURE.md" in context, (
"BackgroundConsciousness._build_context() must include a '## ARCHITECTURE.md' section. "
"the wake-up's Main context must include a '## ARCHITECTURE.md' section. "
"This is a core governance artifact — see docs/DEVELOPMENT.md."
)
assert "This is the architecture doc." in context
def test_consciousness_context_architecture_before_knowledge_base(tmp_path):
"""ARCHITECTURE.md section must come before knowledge base in consciousness context."""
import queue
from unittest.mock import patch, MagicMock
def test_wake_context_architecture_before_knowledge_base(tmp_path):
"""ARCHITECTURE.md section must come before knowledge base in the wake-up's context."""
repo_dir = tmp_path / "repo"
(repo_dir / "docs").mkdir(parents=True)
(repo_dir / "BIBLE.md").write_text("# BIBLE", encoding="utf-8")
(repo_dir / "docs" / "ARCHITECTURE.md").write_text("# ARCH CONTENT", encoding="utf-8")
drive_root = tmp_path / "data"
(drive_root / "logs").mkdir(parents=True)
(drive_root / "state").mkdir(parents=True)
(drive_root / "state" / "state.json").write_text("{}", encoding="utf-8")
(drive_root / "memory").mkdir(parents=True)
kb = drive_root / "memory" / "knowledge"
kb = tmp_path / "data" / "memory" / "knowledge"
kb.mkdir(parents=True)
(kb / "index-full.md").write_text("# Knowledge base index", encoding="utf-8")
eq = queue.Queue()
from ouroboros.consciousness import BackgroundConsciousness
with patch.object(BackgroundConsciousness, "_build_registry", return_value=MagicMock()), \
patch("ouroboros.consciousness.build_memory_sections", return_value=[]), \
patch("ouroboros.consciousness.build_health_invariants", return_value=""), \
patch("ouroboros.consciousness.build_runtime_section", return_value="## Runtime\n\nok"), \
patch("ouroboros.consciousness.build_recent_sections", return_value=[]):
bc = BackgroundConsciousness(
drive_root=drive_root,
repo_dir=repo_dir,
event_queue=eq,
owner_chat_id_fn=lambda: 1,
)
context = bc._build_context()
context = _wake_context(tmp_path)
arch_pos = context.find("## ARCHITECTURE.md")
kb_pos = context.find("## Knowledge base")
assert arch_pos != -1, "ARCHITECTURE.md section not found in consciousness context"
assert arch_pos != -1, "ARCHITECTURE.md section not found in the wake-up's context"
if kb_pos != -1:
assert arch_pos < kb_pos, (
"ARCHITECTURE.md must appear before the knowledge base in consciousness context"
"ARCHITECTURE.md must appear before the knowledge base in the wake-up's context"
)
@ -394,70 +377,39 @@ def test_governance_doc_load_emits_explicit_omission_marker_on_missing(tmp_path)
# Never raises — the function still degrades gracefully.
def test_consciousness_logs_warning_when_architecture_md_missing(tmp_path):
"""BackgroundConsciousness._build_context must log a warning when ARCHITECTURE.md is absent.
def test_wake_context_logs_warning_when_architecture_md_missing(tmp_path, caplog):
"""The Main context builder must log a warning when ARCHITECTURE.md is absent — the
wake-up (an ordinary Main turn) inherits it.
Per the Core Governance Artifacts invariant in docs/DEVELOPMENT.md:
'Log a warning if the file is missing or unavailable — do not silently skip.'
Uses a fresh MagicMock() per getLogger call to avoid mutating real logger singletons.
"""
import queue
from unittest.mock import patch, MagicMock
import logging
repo_dir = tmp_path / "repo"
(repo_dir / "docs").mkdir(parents=True)
(repo_dir / "BIBLE.md").write_text("# BIBLE", encoding="utf-8")
# Deliberately do NOT create docs/ARCHITECTURE.md
drive_root = tmp_path / "data"
(drive_root / "logs").mkdir(parents=True)
(drive_root / "state").mkdir(parents=True)
(drive_root / "state" / "state.json").write_text("{}", encoding="utf-8")
(drive_root / "memory").mkdir(parents=True)
eq = queue.Queue()
from ouroboros.consciousness import BackgroundConsciousness
# A single shared mock logger so all getLogger(name) calls return the same object.
# This avoids mutating any real logger singleton.
mock_logger = MagicMock()
with patch.object(BackgroundConsciousness, "_build_registry", return_value=MagicMock()), \
patch("ouroboros.consciousness.build_memory_sections", return_value=[]), \
patch("ouroboros.consciousness.build_health_invariants", return_value=""), \
patch("ouroboros.consciousness.build_runtime_section", return_value="## Runtime\n\nok"), \
patch("ouroboros.consciousness.build_recent_sections", return_value=[]), \
patch("logging.getLogger", return_value=mock_logger):
bc = BackgroundConsciousness(
drive_root=drive_root,
repo_dir=repo_dir,
event_queue=eq,
owner_chat_id_fn=lambda: 1,
)
context = bc._build_context()
with caplog.at_level(logging.WARNING, logger="ouroboros.context"):
context = _wake_context(tmp_path)
# 1. ARCHITECTURE.md section must be absent from context (file doesn't exist)
assert "## ARCHITECTURE.md" not in context, (
"ARCHITECTURE.md section should not appear when file is missing"
)
# 2. mock_logger.warning must have been called at least once with ARCHITECTURE.md in the message
warning_messages = [
str(call_args)
for call_args in mock_logger.warning.call_args_list
]
arch_warnings = [w for w in warning_messages if "ARCHITECTURE.md" in w]
# 2. a warning naming ARCHITECTURE.md must have been logged
arch_warnings = [record.getMessage() for record in caplog.records
if record.levelno >= logging.WARNING and "ARCHITECTURE.md" in record.getMessage()]
assert arch_warnings, (
"BackgroundConsciousness._build_context() must call logger.warning with 'ARCHITECTURE.md' "
"when the file is missing. Core Governance Artifacts invariant in DEVELOPMENT.md. "
f"All warning calls: {warning_messages}"
"the Main context builder must log a warning naming 'ARCHITECTURE.md' when the file "
"is missing. Core Governance Artifacts invariant in DEVELOPMENT.md. "
f"All records: {[r.getMessage() for r in caplog.records]}"
)
assert any("not found" in w or "empty" in w for w in arch_warnings), (
f"Warning message must indicate the file is missing/empty, got: {arch_warnings}"
assert any("not found" in w or "empty" in w or "unavailable" in w for w in arch_warnings), (
f"Warning message must indicate the file is missing/unavailable, got: {arch_warnings}"
)
def test_development_md_contains_core_governance_invariant():
"""docs/DEVELOPMENT.md must contain the core governance artifact invariant rule."""
import pathlib

6
tests/test_server_shutdown.py
View file

@ -637,6 +637,12 @@ def _supervisor_harness(monkeypatch, tmp_path, steps):
def stop(self):
pass
def tick(self, now=None):
pass
def notify(self, reason):
pass
import time as time_mod
noop = lambda *_a, **_k: None # noqa: E731

12
tests/test_settings_honesty.py
View file

@ -159,6 +159,18 @@ def test_provider_base_url_keys_apply_on_the_next_task_not_restart(monkeypatch,
assert data.get("next_task_changed") is True
def test_consciousness_wake_bounds_apply_without_a_restart(monkeypatch, isolated_settings):
"""The alarm clock reads OUROBOROS_BG_WAKEUP_MIN/MAX through the config getters at each
decision (consciousness.tick / set_next_wakeup), like the other consciousness keys, so
a save never claims a restart for them (Background Consciousness redesign P2)."""
from ouroboros.settings_scales import RESTART_REQUIRED_SETTINGS
assert not {"OUROBOROS_BG_WAKEUP_MIN", "OUROBOROS_BG_WAKEUP_MAX"} & RESTART_REQUIRED_SETTINGS
data = _save(monkeypatch, isolated_settings, {"OUROBOROS_BG_WAKEUP_MIN": "1200", "OUROBOROS_BG_WAKEUP_MAX": "7200"})
assert not data.get("restart_required")
assert not ({"OUROBOROS_BG_WAKEUP_MIN", "OUROBOROS_BG_WAKEUP_MAX"} & set(data.get("restart_keys") or []))
def test_host_service_port_requires_a_restart(monkeypatch, isolated_settings):
"""The host-service port is bound once at server startup."""
data = _save(monkeypatch, isolated_settings, {"OUROBOROS_HOST_SERVICE_PORT": "18999"})

3
tests/test_smoke.py
View file

@ -111,7 +111,8 @@ EXPECTED_TOOLS = [
"peek_task", "discard_child_result", "override_delegation_constraint",
"request_deep_self_review", "chat_history", "update_scratchpad",
"send_user_message", "update_identity", "toggle_evolution",
"toggle_consciousness", "switch_model", "get_task_result",
"toggle_consciousness",
"set_next_wakeup", "switch_model", "get_task_result",
"wait_task", "wait_tasks", "tree_note", "tree_read",
"delegate_start", "delegate_wait", "delegate_cancel", "delegate_answer",
"read_file", "list_files", "write_file", "edit_text",

53
tests/test_transport_death_retry.py
View file

@ -15,7 +15,6 @@ from __future__ import annotations
import json
import queue
from types import SimpleNamespace
from unittest.mock import MagicMock, patch
import httpx
import pytest
@ -675,58 +674,6 @@ def test_default_budget_is_zero_for_every_direct_caller(tmp_path, no_sleep):
assert llm.calls == 1
def test_background_consciousness_never_enters_the_paid_repeat_rail(tmp_path):
"""Owner decision: Background Consciousness gets zero paid transport repeats.
Behavioral contract through one real cycle (`_think_scoped` with the real
`chat_observed`): a client whose send dies with a typed transport death
(httpx ReadError via `__cause__`, capture dispatched) is sent exactly ONCE —
no repeat, no round record in any durable row — and the cycle takes its own
failure path: one `consciousness_llm_error` receipt, idle reason
`llm_error`, acknowledgement withheld, wake-up backoff doubled."""
from ouroboros.consciousness import BackgroundConsciousness
drive_root = tmp_path / "drive"
(drive_root / "logs").mkdir(parents=True)
repo_dir = tmp_path / "repo"
repo_dir.mkdir()
with patch.object(BackgroundConsciousness, "_build_registry", return_value=MagicMock()):
bc = BackgroundConsciousness(
drive_root=drive_root, repo_dir=repo_dir, event_queue=None, owner_chat_id_fn=lambda: None,
)
class _DyingClient:
calls = 0
def _resolve_remote_target(self, _model):
return {"provider": "openrouter"} # Real route shape; local sizing makes no provider call.
def chat(self, **_kwargs):
self.calls += 1
raise _death(state="dispatched")
client = _DyingClient()
bc._llm = client
wakeup_before = bc._next_wakeup_sec
with (
patch.object(bc, "_build_context", return_value="context"),
patch.object(bc, "_tool_schemas", return_value=[]),
patch.object(bc, "_check_budget", return_value=True),
):
assert bc._think_scoped() is False
assert client.calls == 1
events_text = (drive_root / "logs" / "events.jsonl").read_text()
rows = [json.loads(line) for line in events_text.splitlines() if line.strip()]
kinds = [row.get("type") for row in rows]
assert kinds.count("consciousness_llm_error") == 1
assert "Connection error." in next(row for row in rows if row.get("type") == "consciousness_llm_error")["error"]
assert not {"llm_api_error", "llm_non_retryable_same_request", "consciousness_thought"} & set(kinds)
assert TRANSPORT_DEATHS_KEY not in events_text
assert bc._last_idle_reason == "llm_error"
assert bc._cycle_ack_allowed is False
assert bc._next_wakeup_sec == min(wakeup_before * 2, bc._wakeup_max)
def test_classifier_and_review_custody_are_unchanged_by_the_rail():
"""The global classifier still says no-resend for a dispatched death, and
review custody (which consults it) keeps refusing a second paid send."""