One memory in every room: canonical identity/scratchpad writes from project rooms and a cognitive-memory baseline in every Presence ceiling

`update_identity` and `update_scratchpad` no longer no-op in project rooms; they
write the canonical root resolved by `canonical_data_root`, so a forked
execution drive still remembers into the root the next context reads. Every
Presence capability ceiling compiles the constant `COGNITIVE_MEMORY_TOOL_NAMES`
baseline (knowledge read/write/list, scratchpad, identity, chat history) next to
the profile's selections — a selected grant keeps its bindings — so the frozen
ceiling remains the single authority and `registry_core.py` is untouched; the
correspondent gains no tool. CORE_TOOL_NAMES references the same class.
Reflection for project roots stays knowledge-only.
This commit is contained in:
Ouroboros 2026-09-15 00:04:09 +03:00
parent 7a9196162a
commit e559460a1c
10 changed files with 404 additions and 30 deletions

View file

@ -15,6 +15,7 @@ from ouroboros.presence_capabilities import (
PresenceScriptTarget,
PresenceToolTarget,
)
from ouroboros.tool_capabilities import COGNITIVE_MEMORY_TOOL_NAMES
PRESENCE_CEILING_SCHEMA_VERSION = 1
_SHA256_LEN = 64
@ -317,6 +318,19 @@ def build_presence_capability_ceiling(
target.skill_name,
)
)
# The cognitive baseline is part of the ceiling, not of the profile: an
# admitted conversation is still this mind, so it keeps its own memory in
# every room instead of losing what the exchange taught it. It adds no
# authority over settings, delivery, or the filesystem. A name the profile
# already selected keeps THAT grant, because its host-authored argument
# bindings are the exact facts apply_presence_argument_bindings overrides
# the model with; an unselected name arrives with no bindings at all.
selected = {grant.name for grant in tools}
tools.extend(
PresenceToolGrant(name)
for name in sorted(COGNITIVE_MEMORY_TOOL_NAMES)
if name not in selected
)
provisional = PresenceCapabilityCeiling(
skill_name=_text(skill_name, "skill_name"),
skill_content_hash=_sha(skill_content_hash, "skill_content_hash"),

View file

@ -7,7 +7,8 @@ A project-scoped task (an external/workspace task, or one given an explicit
a task's child drive — so it persists across forked/empty runs;
- OUTSIDE ``memory/knowledge/**`` and any ``_copy_stable_memory`` path — so it
never leaks into the forked seed or another project (red-team R3.1/guard #2);
- never identity — there is no per-project identity.
- facts only — identity and the scratchpad stay canonical and are written from
any room through their own tools; there is no per-project copy of either.
This is a thin SSOT helper, NOT a parallel memory subsystem (P7): the existing
knowledge tool + context loader simply redirect their base dir when a task is

View file

@ -13,6 +13,17 @@ OWNER_DELIVERY_TOOL_NAMES: frozenset[str] = frozenset({
"send_user_message", "send_photo", "send_video", "send_file", "send_links",
})
# One class: an actor's own memory. Reading and revising what I know about my
# work and about the people I talk with is a cognitive capability of this mind,
# not an authority over settings, delivery, or anything outside it. Every room
# carries it — the main chat, a project room, and an admitted presence
# conversation, whose ceiling compiles this set in
# ouroboros/presence_authority.py::build_presence_capability_ceiling.
COGNITIVE_MEMORY_TOOL_NAMES: frozenset[str] = frozenset({
"knowledge_read", "knowledge_write", "knowledge_list",
"update_scratchpad", "update_identity", "chat_history",
})
CORE_TOOL_NAMES: frozenset[str] = frozenset({
"read_file", "list_files", "write_file", "edit_text",
"apply_patch", "edit_batch",
@ -41,9 +52,8 @@ CORE_TOOL_NAMES: frozenset[str] = frozenset({
# set today, this makes the coupling explicit).
"list_projects", "route_to_project", "promote_chat_to_task", "steer_task",
"ensure_project_scope",
"update_scratchpad", "update_identity",
"chat_history", "recent_tasks",
"knowledge_read", "knowledge_write", "knowledge_list",
*COGNITIVE_MEMORY_TOOL_NAMES,
"recent_tasks",
"web_search",
"browse_page", "browser_action", "analyze_screenshot", "view_image",
"ocr_pdf", "youtube_transcript", "extract_video_frames",

View file

@ -314,10 +314,11 @@ def get_tools() -> List[ToolEntry]:
ToolEntry("update_scratchpad", {
"name": "update_scratchpad",
"description": "Append a block to your working memory (scratchpad). Each call adds a "
"timestamped block; oldest blocks are auto-evicted when the cap (10) is reached. "
"timestamped block; oldest blocks are auto-evicted when either cap is reached "
"(10 blocks, 60000 characters of content). "
"Write what matters NOW — active tasks, decisions, observations. "
"Persists across sessions, read at every task start. "
"No-op on a project-scoped task (no per-project scratchpad); use knowledge_write for project facts.",
"Project rooms included — the scratchpad is the same working memory in every room.",
"parameters": {"type": "object", "properties": {
"content": {"type": "string", "description": "Content for this scratchpad block"},
}, "required": ["content"]},
@ -342,7 +343,7 @@ def get_tools() -> List[ToolEntry]:
"Use this only after substantive reflection or real experience — not on a "
"greeting or trivial turn. This is the only correct way to write identity; "
"never write memory/identity.md through write_file/edit_text. "
"No-op on a project-scoped task (identity is global and continuous, never per-project).",
"Project rooms included — identity is the same continuous file in every room.",
"parameters": {"type": "object", "properties": {
"content": {"type": "string", "description": "Full identity content (prefer evolving over rewriting from scratch)"},
}, "required": ["content"]},

View file

@ -186,12 +186,6 @@ def _chat_history(
def _update_scratchpad(ctx: ToolContext, content: str) -> str:
"""LLM-driven scratchpad update — appends a timestamped block (Constitution P5: LLM-first)."""
if str(getattr(ctx, "project_id", "") or "").strip():
# Project-scoped tasks have no per-project scratchpad and must never write
# the canonical scratchpad (outbound isolation). Persist project facts via
# knowledge_write instead (routed to the per-project store).
return ("OK: scratchpad is not used for project-scoped tasks (no per-project "
"scratchpad). Persist durable project facts with knowledge_write.")
if not content or not isinstance(content, str) or len(content.strip()) < 10:
return (
_publish_tool_result(ctx, ToolResult(status="error", code="TOOL_ARG_ERROR", text=("⚠️ REJECTED: content is empty or too short "
@ -200,7 +194,14 @@ def _update_scratchpad(ctx: ToolContext, content: str) -> str:
"This likely means the tool call was malformed — check your arguments.")))
)
from ouroboros.memory import Memory
mem = Memory(drive_root=ctx.drive_root)
from ouroboros.tool_access import canonical_data_root
# One working memory, every room (P1): the scratchpad is the same file in
# the main chat, in a project room, and in an external conversation, so a
# project-scoped turn writes it like any other turn. The root follows the
# same precedence as _chat_history, so a forked execution drive still
# remembers into the canonical root the next context reads.
mem = Memory(drive_root=canonical_data_root(ctx))
mem.ensure_files()
try:
block = mem.append_scratchpad_block(
@ -256,11 +257,6 @@ def _send_user_message(ctx: ToolContext, text: str, reason: str = "") -> str:
def _update_identity(ctx: ToolContext, content: str) -> str:
"""Update identity manifest (who you are, who you want to become)."""
if str(getattr(ctx, "project_id", "") or "").strip():
# Identity is global and continuous (P1); it is never modified from a
# project-scoped task. There is no per-project identity.
return ("OK: identity is global and is never modified from a project-scoped "
"task (identity stays continuous across projects — P1).")
if not content or not isinstance(content, str) or len(content.strip()) < 50:
return (
_publish_tool_result(ctx, ToolResult(status="error", code="TOOL_ARG_ERROR", text=("⚠️ REJECTED: content is empty or too short "
@ -269,11 +265,18 @@ def _update_identity(ctx: ToolContext, content: str) -> str:
"This likely means the tool call was malformed — check your arguments.")))
)
from ouroboros.memory import Memory
mem = Memory(drive_root=ctx.drive_root)
from ouroboros.tool_access import canonical_data_root
# One identity, every room (P1): who I am does not change with the room I
# am speaking in, so a project room or an external conversation revises the
# same continuous file. The root follows the same precedence as
# _chat_history, so a forked execution drive still writes the identity the
# canonical root reads back.
mem = Memory(drive_root=canonical_data_root(ctx))
mem.ensure_files()
old_content = ""
path = ctx.drive_root / "memory" / "identity.md"
path = mem.identity_path()
if path.exists():
try:
old_content = path.read_text(encoding="utf-8")

View file

@ -0,0 +1,105 @@
"""One memory across rooms: identity and scratchpad are the same files anywhere.
A project room is a focused working room, not a second mind (BIBLE P1). These
tests pin the behaviour the tools give the model: a project-scoped turn revises
the canonical identity and scratchpad exactly as the main chat does, and a
forked execution drive still writes into the root the next context reads back.
"""
from __future__ import annotations
import json
from ouroboros.tools import control_runtime
from ouroboros.tools.registry import ToolContext
_NOTE = "a meaningful scratchpad note written from inside a project room"
_IDENTITY = (
"I am Ouroboros. I keep one continuous self across every room I speak in, "
"and I revise this file when experience genuinely changes it."
)
def _ctx(drive_root, **kwargs) -> ToolContext:
return ToolContext(
repo_dir=drive_root.parent / "repo",
drive_root=drive_root,
task_id="t-unified",
**kwargs,
)
def _blocks(drive_root):
path = drive_root / "memory" / "scratchpad_blocks.json"
return json.loads(path.read_text(encoding="utf-8")) if path.exists() else []
def test_project_room_writes_the_canonical_scratchpad(tmp_path):
data = tmp_path / "data"
data.mkdir()
result = control_runtime._update_scratchpad(_ctx(data, project_id="proj_p"), _NOTE)
assert result.startswith("OK: scratchpad block appended")
blocks = _blocks(data)
assert [block["content"] for block in blocks] == [_NOTE]
def test_project_room_writes_the_canonical_identity(tmp_path):
data = tmp_path / "data"
data.mkdir()
result = control_runtime._update_identity(_ctx(data, project_id="proj_p"), _IDENTITY)
assert result.startswith("OK: identity updated")
assert (data / "memory" / "identity.md").read_text(encoding="utf-8") == _IDENTITY
journal = (data / "memory" / "identity_journal.jsonl").read_text(encoding="utf-8")
assert json.loads(journal.strip().splitlines()[-1])["new_content"] == _IDENTITY
def test_main_chat_writes_the_same_files(tmp_path):
# The room changes nothing: an unscoped turn lands in exactly one place.
data = tmp_path / "data"
data.mkdir()
ctx = _ctx(data)
control_runtime._update_scratchpad(ctx, _NOTE)
control_runtime._update_identity(ctx, _IDENTITY)
assert [block["content"] for block in _blocks(data)] == [_NOTE]
assert (data / "memory" / "identity.md").read_text(encoding="utf-8") == _IDENTITY
def test_forked_execution_drive_remembers_into_the_canonical_root(tmp_path):
# A forked task executes on its own drive, but memory belongs to the root
# the next context reads — the same precedence chat_history already uses.
canonical = tmp_path / "data"
forked = tmp_path / "fork"
for path in (canonical, forked):
path.mkdir()
ctx = _ctx(
forked,
project_id="proj_p",
task_metadata={"budget_drive_root": str(canonical)},
)
control_runtime._update_scratchpad(ctx, _NOTE)
control_runtime._update_identity(ctx, _IDENTITY)
assert [block["content"] for block in _blocks(canonical)] == [_NOTE]
assert (canonical / "memory" / "identity.md").read_text(encoding="utf-8") == _IDENTITY
assert not (forked / "memory" / "identity.md").exists()
assert not (forked / "memory" / "scratchpad_blocks.json").exists()
def test_context_budget_root_is_used_when_metadata_is_absent(tmp_path):
canonical = tmp_path / "data"
forked = tmp_path / "fork"
for path in (canonical, forked):
path.mkdir()
ctx = _ctx(forked, budget_drive_root=str(canonical))
control_runtime._update_scratchpad(ctx, _NOTE)
assert [block["content"] for block in _blocks(canonical)] == [_NOTE]
assert not (forked / "memory" / "scratchpad_blocks.json").exists()

View file

@ -134,7 +134,16 @@ def test_admission_freezes_reviewed_behavior_runtime_digests_and_authority(tmp_p
assert admission.origin == binding.origin
assert admission.destination == binding.destination
assert admission.capability_ceiling.skill_name == "community-helper"
assert [grant.name for grant in admission.capability_ceiling.tool_grants] == ["chat_history"]
# The reviewed profile selected chat_history; the rest is the constant
# cognitive baseline every admitted conversation carries.
assert [grant.name for grant in admission.capability_ceiling.tool_grants] == [
"chat_history",
"knowledge_list",
"knowledge_read",
"knowledge_write",
"update_identity",
"update_scratchpad",
]
assert admission.capability_ceiling.skill_content_hash == admission.skill_content_hash
assert admission.capability_ceiling.profile_fingerprint == admission.profile_fingerprint
assert admission.capability_ceiling.state_fingerprint == admission.state_fingerprint

View file

@ -58,7 +58,17 @@ def test_ceiling_compiles_exact_tools_scripts_resources_and_digest():
),
)
assert [grant.name for grant in ceiling.tool_grants] == ["chat_history", "skill_exec"]
# The profile selected chat_history and one script; the cognitive baseline
# (own memory, no new authority) is compiled in beside them.
assert [grant.name for grant in ceiling.tool_grants] == [
"chat_history",
"knowledge_list",
"knowledge_read",
"knowledge_write",
"skill_exec",
"update_identity",
"update_scratchpad",
]
script = next(grant for grant in ceiling.tool_grants if grant.name == "skill_exec")
assert [(item.argument_path, item.static_value) for item in script.bindings] == [
(("skill",), "calendar"),
@ -215,10 +225,35 @@ def test_registry_filters_schema_dispatch_and_resolved_targets(tmp_path):
registry.set_context(ctx)
names = {schema["function"]["name"] for schema in registry.schemas()}
assert names == {"presence_finish", "presence_cancel_work", "read_file"}
assert names == {
"presence_finish",
"presence_cancel_work",
"read_file",
"chat_history",
"knowledge_list",
"knowledge_read",
"knowledge_write",
"update_identity",
"update_scratchpad",
}
# Own memory is advertised and it runs; nothing that acts outside this mind
# comes with it, in the schemas or in dispatch.
assert "PRESENCE_CAPABILITY_BLOCKED" not in registry.execute(
"knowledge_write",
{"topic": "presence-note", "content": "what this exchange taught me", "scope": "global"},
)
assert (data / "memory" / "knowledge" / "presence-note.md").exists()
for blocked in ("write_file", "run_command", "send_user_message"):
assert blocked not in names
assert "PRESENCE_CAPABILITY_BLOCKED" in registry.execute(
"run_command", {"command": "pwd"}
)
assert "PRESENCE_CAPABILITY_BLOCKED" in registry.execute(
"write_file", {"root": "active_workspace", "path": "new.txt", "content": "no"}
)
assert "PRESENCE_CAPABILITY_BLOCKED" in registry.execute(
"send_user_message", {"text": "no"}
)
assert "PRESENCE_RESOURCE_BLOCKED" in registry.execute(
"read_file", {"root": "active_workspace", "path": "private.txt"}
)

View file

@ -0,0 +1,195 @@
"""An admitted presence conversation keeps its own memory, and nothing more.
The ceiling is compiled from the reviewed profile plus one constant set: the
tools by which this mind reads and revises what it knows
(`tool_capabilities.COGNITIVE_MEMORY_TOOL_NAMES`). The baseline adds no
authority to act outside the mind, it is covered by the ceiling digest, and it
never displaces a selection the profile authored with argument bindings.
"""
from __future__ import annotations
import json
import pytest
from ouroboros.presence_authority import (
PresenceAuthorityError,
apply_presence_argument_bindings,
build_presence_capability_ceiling,
presence_ceiling_allows_tool,
presence_ceiling_from_payload,
presence_ceiling_payload,
)
from ouroboros.presence_capabilities import (
PresenceArgumentBinding,
PresenceProfileResolution,
PresenceSelection,
PresenceToolTarget,
)
from ouroboros.presence_runtime import ResolvedPresenceRuntime
from ouroboros.tool_capabilities import COGNITIVE_MEMORY_TOOL_NAMES
from ouroboros.tools.registry import ToolContext
def _resolution(*selections):
return PresenceProfileResolution(
active=tuple(selections),
missing_required=(),
missing_optional=(),
orphaned=(),
runtime=ResolvedPresenceRuntime("main", 10, 10, False),
profile_fingerprint="a" * 64,
selection_fingerprint="b" * 64,
required_selections_present=True,
)
def _ceiling(*selections):
return build_presence_capability_ceiling(
skill_name="community-helper",
skill_content_hash="c" * 64,
state_fingerprint="d" * 64,
resolution=_resolution(*selections),
)
def test_profile_without_tool_selections_still_carries_its_own_memory():
ceiling = _ceiling()
assert [grant.name for grant in ceiling.tool_grants] == sorted(COGNITIVE_MEMORY_TOOL_NAMES)
assert all(grant.bindings == () for grant in ceiling.tool_grants)
for name in COGNITIVE_MEMORY_TOOL_NAMES:
assert presence_ceiling_allows_tool(ceiling, name)
def test_the_baseline_grants_no_authority_outside_the_mind():
ceiling = _ceiling()
for name in ("write_file", "edit_text", "run_command", "send_user_message", "skill_exec"):
assert not presence_ceiling_allows_tool(ceiling, name)
def test_a_selected_baseline_tool_keeps_the_profile_authored_bindings():
# Deduplication is by name and the profile wins: its bindings are exact
# host facts, and losing them would hand the argument back to the model.
selection = PresenceSelection(
"1" * 64,
PresenceToolTarget("builtin", "knowledge_write"),
(PresenceArgumentBinding(("scope",), "static", static_value="global"),),
)
ceiling = _ceiling(selection)
grant = next(item for item in ceiling.tool_grants if item.name == "knowledge_write")
assert [(item.argument_path, item.static_value) for item in grant.bindings] == [
(("scope",), "global"),
]
assert [item.name for item in ceiling.tool_grants] == sorted(COGNITIVE_MEMORY_TOOL_NAMES)
ctx = ToolContext(
repo_dir=None,
drive_root=None,
task_contract={"capability_ceiling": presence_ceiling_payload(ceiling)},
)
bound = apply_presence_argument_bindings(
ctx, "knowledge_write", {"topic": "note", "scope": "project:sneaky"},
)
assert bound == {"topic": "note", "scope": "global"}
# A baseline name the profile did not select arrives unbound: the model
# supplies its own arguments, exactly as in any other room.
assert apply_presence_argument_bindings(
ctx, "knowledge_read", {"topic": "note"},
) == {"topic": "note"}
def test_the_digest_covers_the_baseline(tmp_path):
ceiling = _ceiling()
payload = presence_ceiling_payload(ceiling)
assert presence_ceiling_from_payload(payload) == ceiling
stripped = json.loads(json.dumps(payload))
stripped["tools"] = [tool for tool in stripped["tools"] if tool["name"] != "update_identity"]
with pytest.raises(PresenceAuthorityError) as caught:
presence_ceiling_from_payload(stripped)
assert caught.value.code == "presence_authority_digest_mismatch"
def test_admitted_external_turn_writes_global_knowledge_and_nothing_else(tmp_path):
"""The whole path, no model: admission → runner → registry → the note on disk.
A real presence admission (reviewed skill, saved selection, bound room) runs
one bounded turn whose agent writes what it learned about a person. The note
lands on the canonical global shelf, the six memory tools are offered, and
every tool that would act outside this mind is absent and refused.
"""
from tests.test_presence_admission import _admit, _binding, _install_behavior, _select_history
from tests.test_presence_runner import _event
from ouroboros.presence_runner import PresenceTurnGate, run_presence_turn
from ouroboros.tools.registry import ToolRegistry
repo = tmp_path / "repo"
data = tmp_path / "data"
repo.mkdir()
data.mkdir()
skill_dir = _install_behavior(data)
_select_history(data, skill_dir)
admission = _admit(data, _binding(data))
assert [grant.name for grant in admission.capability_ceiling.tool_grants] == sorted(
COGNITIVE_MEMORY_TOOL_NAMES
)
seen: dict[str, object] = {}
class Agent:
def __init__(self, repo_dir, drive_root, **_kwargs):
self.repo_dir = repo_dir
self.drive_root = drive_root
def handle_task(self, task):
ctx = ToolContext(
repo_dir=self.repo_dir,
drive_root=self.drive_root,
task_id=str(task.get("id") or "presence-turn"),
task_contract=task.get("task_contract") or {},
task_metadata=task.get("metadata") or {},
current_chat_id=task.get("chat_id"),
)
registry = ToolRegistry(repo_dir=self.repo_dir, drive_root=self.drive_root)
registry.set_context(ctx)
seen["schemas"] = {schema["function"]["name"] for schema in registry.schemas()}
seen["write"] = registry.execute(
"knowledge_write",
{
"topic": "people/alex",
"scope": "global",
"content": "Alex asked for short answers today; I read it as a preference to test.",
},
)
seen["refused"] = {
name: registry.execute(name, args)
for name, args in (
("write_file", {"root": "runtime_data", "path": "memory/identity.md", "content": "no"}),
("send_user_message", {"text": "no"}),
("run_command", {"cmd": ["true"]}),
)
}
return [{"type": "presence_result", "outcome": "message", "text": "Noted.", "work_ref": ""}]
result = run_presence_turn(
admission=admission,
event=_event(),
repo_dir=repo,
drive_root=data,
agent_factory=lambda repo_dir, drive_root, **kwargs: Agent(repo_dir, drive_root, **kwargs),
gate=PresenceTurnGate(2),
)
assert result.outcome == "message"
note = (data / "memory" / "knowledge" / "people" / "alex.md").read_text(encoding="utf-8")
assert "short answers" in note
assert "PRESENCE_CAPABILITY_BLOCKED" not in str(seen["write"])
assert COGNITIVE_MEMORY_TOOL_NAMES <= seen["schemas"]
for name, refusal in seen["refused"].items():
assert name not in seen["schemas"]
assert "PRESENCE_CAPABILITY_BLOCKED" in refusal

View file

@ -328,19 +328,20 @@ def test_scheduled_subagent_task_inherits_project_id():
assert resolve_project_id(task) == "proj_x"
def test_scratchpad_and_identity_tools_noop_for_project_tasks(tmp_path):
def test_scratchpad_and_identity_tools_write_canonical_memory_from_a_project_room(tmp_path):
import types
from ouroboros.tools import control
# One memory, every room: a project-scoped turn writes the same canonical
# scratchpad and identity files the main chat writes.
ctx = types.SimpleNamespace(drive_root=tmp_path, project_id="proj_p")
r1 = control._update_scratchpad(ctx, "a meaningful scratchpad note for the task at hand")
r2 = control._update_identity(ctx, "x" * 60)
assert "project-scoped" in r1.lower()
assert ("project-scoped" in r2.lower()) or ("global" in r2.lower())
# nothing written to canonical memory
assert not (tmp_path / "memory" / "scratchpad_blocks.json").exists()
assert not (tmp_path / "memory" / "identity.md").exists()
assert r1.startswith("OK")
assert r2.startswith("OK")
assert (tmp_path / "memory" / "scratchpad_blocks.json").exists()
assert (tmp_path / "memory" / "identity.md").read_text(encoding="utf-8") == "x" * 60
def test_maybe_promote_skips_project_scoped_task(tmp_path, monkeypatch):