from __future__ import annotations
import copy
import hashlib
import inspect
import io
import json
import os
import shutil
import subprocess
import sys
import tomllib
import urllib.parse
import uuid
from contextlib import AbstractContextManager
from dataclasses import asdict
from http.client import HTTPMessage
from pathlib import Path
from types import SimpleNamespace
from typing import Any, Literal, Protocol, cast
from unittest import mock
import pytest
import torch
import torch.nn.functional as F
from safetensors import safe_open as _safe_open_untyped
from safetensors.torch import load_file, save_file
from nexum_core import (
NexumCoreConfig,
bind_full_state,
load_token_bridge,
observe_tool_outcomes as observe_tool_outcomes_api,
wire_full_authority,
)
from nexum_core.capacity_stack import (
CapacityLMHead,
NativeDecodeConfidenceSurface,
_retained_episode_route,
_retained_evidence_route,
_retained_token_logit_update,
native_decode_confidence_training_loss,
)
from nexum_core.config import NexumConfig
from nexum_core.experts import FactorizedGranularExpertBank, MHCExpertBank
from nexum_core.internal_agents import (
merge_selected_agent_workspaces,
run_internal_agent_workspaces,
)
from nexum_core.numbered_io import open_tensor_package
from nexum_core.none_expert_paging import (
NoNEActivePageParameters,
NoNECapacityPageStoreBoundary,
NoNEFrozenExpertCatalogBoundary,
NoNEFrozenPageBatch,
NoNEPageRequestPacket,
NoNEPagedExpertCompute,
)
from nexum_core.none_paged_layer import (
NexumNoNEFrozenRouterLayer,
NexumNoNEPagedLayerBoundary,
NexumNoNEPagedLayerTransaction,
NexumNoNEPagedStack,
_route_request_index,
)
from nexum_core.outcome_history import ensure_outcome_history
from nexum_core.rbo import (
STOP_REASON_ROUTE_ARMS_EXHAUSTED,
NexumRBOConfig,
NexumRBOResult,
)
from nexum_core.routing import ensure_domain_router
from nexum_core.self_correction import (
SelfCorrectionTriggerBank,
ensure_self_correction_bank,
)
from nexum_core.token_tbr import NexumTBR, native_bit_code
from nexum_core.tool_outcomes import observe_tool_outcomes
from nexum_core.traversal_registry import TraversalPathRegistry
from nexum_runtime.architecture import NexumAttention, NexumExperts, NexumForCausalLM
from nexum_runtime.architecture_config import NexumArchitectureConfig
from nexum_runtime.bundle import BundleReport, release_artifact_sha256, validate_bundle
from nexum_runtime.cli import _validated_candidate_identity
from nexum_runtime.correction import inspect_tool_text
from nexum_runtime.executor import (
TOOL_CALL_END,
TOOL_CALL_START,
execute_tool_calls,
execute_tool_text,
list_tools,
parse_tool_calls,
tool_schemas,
_public_http_response,
)
from nexum_runtime.harness import (
NNFXHarnessRequest,
NNFXHarnessSession,
_accumulate_runtime_trajectory,
)
from nexum_runtime.runner import (
NexumLocalRunner,
_NativeSelectedTokenRecorder,
_action_package_contract,
_architecture_config,
_authority_geometry_from_package,
_authority_package_contract,
_context_intent_package_contract,
_context_intent_action_readiness_probe,
_current_failure_action_evidence,
_full_generation_engagement,
_incremental_uncapped_generate,
_link_or_copy,
_load_numbered_action,
_load_numbered_context_intent,
_load_numbered_primary,
_load_tokenizer,
_grounded_observation_token_evidence,
_materialize_runtime_buffers,
_merge_numbered_state,
_native_confidence_checkpoint_ready,
_stopping_criteria_resolved,
_native_tool_call_stopping_criteria,
_normalize_messages,
_numbered_none_page_view,
_observe_grounded_tool_receipts,
_prepare_generation_context,
_protocol_rejection_result,
_prompt_from_messages,
_required_tensor_path,
_runtime_state_root,
_shared_learning_validation,
_tool_contract_fingerprints_t,
_tool_protocol_boundary,
_token_id_fingerprint_t,
_trusted_polarized_action_evidence,
_validate_numbered_state_targets,
_wire_tool_calls,
runtime_state_namespace,
)
from nexum_runtime.session_state import NexumSessionStateBank
from nexum_runtime.server import (
NexumHandler,
_responses_caller_observations,
_responses_chat,
_responses_chat_payload,
_responses_result,
_session_bound_chat_result,
_visible_chat_result,
serve,
)
from nexum_runtime.status import runtime_health
from nexum_runtime.templates import render_template, template_names
from nexum_runtime.tooling.contracts import ToolCall, ToolExecutionResult
from nexum_runtime.visibility import (
private_reasoning_content,
private_reasoning_only,
visible_completion,
)
RELEASE_ROOT = Path(__file__).resolve().parents[2]
MODEL_DIR = RELEASE_ROOT / "model"
RUNTIME_SRC = RELEASE_ROOT / "runtime" / "src"
def test_candidate_identity_rejects_blank_model_directory() -> None:
with pytest.raises(ValueError, match="model directory is required"):
_validated_candidate_identity(" ")
def _source_cli_environment() -> dict[str, str]:
environment = dict(os.environ)
current = environment.get("PYTHONPATH", "")
source = str(RUNTIME_SRC)
environment["PYTHONPATH"] = (
source if not current else f"{source}{os.pathsep}{current}"
)
return environment
class _SafeTensorHandle(Protocol):
def keys(self) -> list[str]: ...
def metadata(self) -> dict[str, str] | None: ...
def get_tensor(self, name: str) -> torch.Tensor: ...
class _SafeOpen(Protocol):
def __call__(
self,
filename: str,
*,
framework: Literal["pt"],
device: str,
) -> AbstractContextManager[_SafeTensorHandle]: ...
_SAFE_OPEN = cast(_SafeOpen, _safe_open_untyped)
def _completion(message: dict[str, object]) -> dict[str, object]:
return {
"id": "chatcmpl-test",
"object": "chat.completion",
"model": "Nexum",
"choices": [{"index": 0, "message": message, "finish_reason": "stop"}],
"usage": {"prompt_tokens": 4, "completion_tokens": 2, "total_tokens": 6},
}
def _tool_call(
call_id: str, name: str, arguments: dict[str, object]
) -> dict[str, object]:
return {
"id": call_id,
"type": "function",
"function": {"name": name, "arguments": json.dumps(arguments)},
}
def test_visible_completion_preserves_private_trace_off_surface() -> None:
raw = "inspect privatelyVerified result."
assert visible_completion(raw) == "Verified result."
assert private_reasoning_content(raw) == "inspect privately"
assert private_reasoning_only("continue checking") is True
assert visible_completion("Visibleprivate tail") == "Visible"
assert private_reasoning_content("Visibleprivate tail") == "private tail"
def test_visible_chat_result_copies_and_strips_private_reasoning() -> None:
raw = "inspect privatelyVerified result."
chat = _completion({"role": "assistant", "content": raw})
original = copy.deepcopy(chat)
projected = _visible_chat_result(cast(dict[str, Any], chat))
source_choices = cast(list[dict[str, object]], chat["choices"])
projected_choices = cast(list[dict[str, object]], projected["choices"])
source_message = cast(dict[str, object], source_choices[0]["message"])
projected_message = cast(dict[str, object], projected_choices[0]["message"])
assert projected is not chat
assert projected_choices is not source_choices
assert projected_message is not source_message
assert projected_message["content"] == "Verified result."
assert projected_message["reasoning_content"] == "inspect privately"
assert chat == original
assert source_message["content"] == raw
def test_session_bound_chat_result_preserves_signed_rejection_receipt() -> None:
signed_observation = {
"name": "MisspelledTool",
"args": {"path": "state.txt"},
"ok": False,
"executed": False,
"tool_call_id": "call_rejected_exact",
"receipt_source": "runtime_rejection",
"receipt_nonce": "rejection-nonce",
"receipt_auth": "rejection-auth",
}
chat = {
"choices": [{"message": {"role": "assistant", "content": None}}],
"nexum": {
"tool_action_protocol_reason": "unknown_tool",
"tool_action_protocol_rejection_observations": [signed_observation],
},
}
projected = _session_bound_chat_result(chat, session_id="session-rejection")
assert projected["nexum"]["session_id"] == "session-rejection"
assert (
projected["nexum"]["tool_action_protocol_rejection_observations"][0]
== signed_observation
)
assert "session_id" not in chat["nexum"]
def test_message_normalization_maps_reasoning_to_tokenizer_thinking() -> None:
normalized = _normalize_messages(
[
{
"role": "assistant",
"content": "Verified result.",
"reasoning_content": "inspect privately",
}
]
)
assert normalized == [
{
"role": "assistant",
"content": "Verified result.",
"thinking": "inspect privately",
}
]
def test_tool_smoke_executes_exact_terminal_action(tmp_path: Path) -> None:
call = f"{TOOL_CALL_START}[Bash(command='printf ok')]{TOOL_CALL_END}"
results = execute_tool_text(call, cwd=str(tmp_path), timeout_s=5)
assert len(results) == 1
assert results[0].ok is True
assert results[0].output == "ok"
assert results[0].args == {"command": "printf ok"}
def test_release_artifact_identity_changes_with_validated_bundle_bytes(
tmp_path: Path,
) -> None:
model_dir = tmp_path / "model"
model_dir.mkdir()
for name in (
"config.json",
"tokenizer.json",
"tokenizer_config.json",
"chat_template.jinja",
"generation_config.json",
"state_config.json",
):
(model_dir / name).write_text("{}", encoding="utf-8")
(model_dir / "tensor_map.json").write_text(
json.dumps(
{
"tensors": [
{
"id": "000001",
"file": "safetensors/000001.safetensors",
"bytes": 8,
"sha256": "a" * 64,
"binding_sha256": "b" * 64,
"descriptor_sha256": "c" * 64,
"key_count": 1,
}
]
}
),
encoding="utf-8",
)
report = BundleReport(
ok=True,
model_dir=str(model_dir),
files=(),
missing=(),
tensor_map_ok=True,
load_map_ok=True,
tensor_count=1,
tensor_bytes=8,
tensor_hashes_verified=True,
tokenizer_ok=True,
errors=(),
)
before = release_artifact_sha256(model_dir, validated_report=report)
(model_dir / "config.json").write_text('{"revision":2}', encoding="utf-8")
after = release_artifact_sha256(model_dir, validated_report=report)
assert len(before) == 64
assert before != after
def test_container_dependencies_match_runtime_package_contract() -> None:
project = tomllib.loads(
(RELEASE_ROOT / "runtime/pyproject.toml").read_text(encoding="utf-8")
)["project"]
expected = [
dependency
for dependency in project["dependencies"]
if not dependency.startswith("torch==")
]
expected.extend(project["optional-dependencies"]["browser"])
actual = [
line.strip()
for line in (RELEASE_ROOT / "runtime/container-requirements.txt")
.read_text(encoding="utf-8")
.splitlines()
if line.strip() and not line.lstrip().startswith("#")
]
assert actual == expected
def test_state_directory_is_bound_to_one_namespace(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.setenv("NEXUM_STATE_DIR", str(tmp_path / "state"))
monkeypatch.setenv("NEXUM_STATE_NAMESPACE", "suite:mode:run-a")
assert runtime_state_namespace() == "suite:mode:run-a"
assert _runtime_state_root() == (tmp_path / "state").resolve()
monkeypatch.setenv("NEXUM_STATE_NAMESPACE", "suite:mode:run-b")
with pytest.raises(RuntimeError, match="another namespace"):
_runtime_state_root()
def test_runtime_view_reuses_complete_copy_and_replaces_stale_copy(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
source = tmp_path / "source" / "row.safetensors"
destination = tmp_path / "cache" / "row.safetensors"
source.parent.mkdir()
source.write_bytes(b"initial")
source_stat = source.stat()
source_mtime_ns = source_stat.st_mtime_ns - 1_000_000_000
os.utime(source, ns=(source_stat.st_atime_ns, source_mtime_ns))
real_copy2 = shutil.copy2
def cross_device_link(_source: Path, _destination: Path) -> None:
raise OSError("cross-device link")
monkeypatch.setattr(os, "link", cross_device_link)
_link_or_copy(source, destination)
assert destination.read_bytes() == b"initial"
assert destination.stat().st_mtime_ns == source_mtime_ns
def reject_recopy(_source: Path, _destination: Path) -> None:
raise AssertionError("matching runtime-view file was recopied")
monkeypatch.setattr(shutil, "copy2", reject_recopy)
_link_or_copy(source, destination)
monkeypatch.setattr(shutil, "copy2", real_copy2)
source.write_bytes(b"updated")
source_stat = source.stat()
updated_mtime_ns = source_mtime_ns + 2_000_000_000
os.utime(source, ns=(source_stat.st_atime_ns, updated_mtime_ns))
_link_or_copy(source, destination)
assert destination.read_bytes() == b"updated"
assert destination.stat().st_mtime_ns == updated_mtime_ns
assert not tuple(destination.parent.glob(f".{destination.name}.*.tmp"))
def test_session_state_pristine_tracks_persisted_activity(tmp_path: Path) -> None:
bank = object.__new__(NexumSessionStateBank)
bank.state_root = tmp_path / "sessions"
bank.learning_root = tmp_path / "learning"
bank.state_root.mkdir()
bank.learning_root.mkdir()
bank._active_session = None
bank._sessions = {}
bank._learning_generation = 0
assert bank.state_pristine is True
(bank.state_root / "activity").write_text("1", encoding="ascii")
assert bank.state_pristine is False
def test_terminal_action_runs_inside_workspace_only_namespace(tmp_path: Path) -> None:
command = (
'test "$PWD" = /workspace && '
"test ! -e /root && test ! -e /etc/passwd && "
"test -x /usr/bin/python3 && "
'test -z "${NEXUM_API_KEY:-}" && '
"printf isolated > namespace-proof.txt"
)
result = execute_tool_text(
f"Bash(command={command!r})", cwd=str(tmp_path), timeout_s=5
)[0]
assert result.ok is True
assert (tmp_path / "namespace-proof.txt").read_text(encoding="utf-8") == "isolated"
def test_published_bare_tool_example_executes(tmp_path: Path) -> None:
results = execute_tool_text(
"Write(path='example.txt', content='published-contract')",
cwd=str(tmp_path),
timeout_s=5,
)
assert len(results) == 1
assert results[0].ok is True
assert (tmp_path / "example.txt").read_text(
encoding="utf-8"
) == "published-contract"
def test_tools_list_has_core_surfaces() -> None:
names = {tool["name"] for tool in list_tools()}
assert {
"Bash",
"Read",
"Write",
"Edit",
"Glob",
"Grep",
"WebFetch",
"ToolCatalog",
"CreateTool",
"RetireTool",
"RunDynamicTool",
} <= names
def test_dynamic_tool_registration_is_distinct_from_execution(tmp_path: Path) -> None:
registered = execute_tool_text(
"CreateTool(name='write_proof', command='printf exact > proof.txt')",
cwd=str(tmp_path),
timeout_s=5,
)[0]
assert registered.ok is True
registration = json.loads(registered.output)
assert registration["registered"] is True
assert registration["dynamic_command_executed"] is False
assert not (tmp_path / "proof.txt").exists()
executed = execute_tool_text(
"RunDynamicTool(name='write_proof')",
cwd=str(tmp_path),
timeout_s=5,
)[0]
assert executed.ok is True
assert (tmp_path / "proof.txt").read_text(encoding="utf-8") == "exact"
def test_harness_executes_model_selected_persistent_tool_lifecycle(
tmp_path: Path,
) -> None:
(tmp_path / "rows.txt").write_text("one\ntwo\n", encoding="utf-8")
payloads: list[dict[str, object]] = []
def complete(payload: dict[str, object]) -> dict[str, object]:
payloads.append(payload)
if len(payloads) == 1:
return _completion(
{
"role": "assistant",
"content": None,
"tool_calls": [
_tool_call(
"catalog-call",
"ToolCatalog",
{"query": "workspace line counter"},
)
],
}
)
if len(payloads) == 2:
return _completion(
{
"role": "assistant",
"content": None,
"tool_calls": [
_tool_call(
"create-call",
"CreateTool",
{
"name": "workspace_line_count",
"command": "wc -l",
"description": "Count lines in a workspace file.",
},
)
],
}
)
if len(payloads) == 3:
return _completion(
{
"role": "assistant",
"content": None,
"tool_calls": [
_tool_call(
"run-call",
"RunDynamicTool",
{"name": "workspace_line_count", "args": "rows.txt"},
)
],
}
)
return _completion(
{"role": "assistant", "content": "The file contains two lines."}
)
response = NNFXHarnessSession(
complete=complete,
sign_observation=lambda _session_id, row: {
**row,
"receipt_nonce": uuid.uuid4().hex,
"receipt_auth": "test-auth",
},
).run(
NNFXHarnessRequest(
prompt="Count the lines using a reusable capability.",
workspace=str(tmp_path),
)
)
assert response.final_text == "The file contains two lines."
assert [row["name"] for row in response.tool_results] == [
"ToolCatalog",
"CreateTool",
"RunDynamicTool",
]
assert response.tool_results[-1]["output"].strip().startswith("2 ")
persisted = tmp_path / ".nexum" / "tools" / "current"
assert (persisted / "workspace_line_count.json").is_file()
first_messages = cast(list[dict[str, object]], payloads[0]["messages"])
assert "When a required capability is missing" in str(first_messages[0]["content"])
assert (
"Before a consequential or terminal action, draft the intended action "
"privately and verify every explicit user constraint against current tool evidence."
in str(first_messages[0]["content"])
)
def test_workspace_relative_file_tools(tmp_path: Path) -> None:
calls = "\n".join(
[
f"{TOOL_CALL_START}[Write(path='src/value.txt', content='old')]{TOOL_CALL_END}",
f"{TOOL_CALL_START}[Read(path='src/value.txt')]{TOOL_CALL_END}",
f"{TOOL_CALL_START}[Edit(path='src/value.txt', old_string='old', new_string='new')]{TOOL_CALL_END}",
f"{TOOL_CALL_START}[Glob(pattern='src/*.txt')]{TOOL_CALL_END}",
]
)
results = execute_tool_text(calls, cwd=str(tmp_path), timeout_s=5)
assert [result.ok for result in results] == [True, True, True, True]
assert (tmp_path / "src" / "value.txt").read_text(encoding="utf-8") == "new"
assert results[0].output == "src/value.txt"
assert results[2].output == "src/value.txt"
assert results[3].output == "src/value.txt"
assert str(tmp_path) not in "\n".join(result.output for result in results)
def test_search_and_registration_outputs_are_workspace_relative(tmp_path: Path) -> None:
(tmp_path / "src").mkdir()
(tmp_path / "src" / "value.txt").write_text("needle\n", encoding="utf-8")
grep = execute_tool_text(
"Grep(pattern='needle', path='src')", cwd=str(tmp_path), timeout_s=5
)[0]
registered = execute_tool_text(
"CreateTool(name='show_value', command='cat src/value.txt')",
cwd=str(tmp_path),
timeout_s=5,
)[0]
assert grep.ok is True
assert grep.output == "src/value.txt:1:needle\n"
assert str(tmp_path) not in grep.output
registration = json.loads(registered.output)
assert registration["path"] == ".nexum/tools/current/show_value.json"
assert str(tmp_path) not in registered.output
def test_dynamic_tool_lifecycle_is_versioned_and_tamper_evident(
tmp_path: Path,
) -> None:
created = execute_tool_text(
"CreateTool(name='versioned_tool', command='printf one')",
cwd=str(tmp_path),
timeout_s=5,
)[0]
assert created.ok is True
creation = json.loads(created.output)
assert creation["generation"] == 1
first_sha256 = creation["definition_sha256"]
duplicate = execute_tool_text(
"CreateTool(name='versioned_tool', command='printf replaced')",
cwd=str(tmp_path),
timeout_s=5,
)[0]
assert duplicate.ok is False
assert "already exists" in duplicate.error
described = execute_tool_text(
"ToolDescribe(name='versioned_tool')",
cwd=str(tmp_path),
timeout_s=5,
)[0]
assert described.ok is True
assert json.loads(described.output)["definition_sha256"] == first_sha256
stale = execute_tool_text(
"UpgradeTool(name='versioned_tool', command='printf two', expected_sha256='0')",
cwd=str(tmp_path),
timeout_s=5,
)[0]
assert stale.ok is False
assert "changed before mutation" in stale.error
upgraded = execute_tool_text(
"UpgradeTool(name='versioned_tool', command='printf two', "
f"expected_sha256='{first_sha256}')",
cwd=str(tmp_path),
timeout_s=5,
)[0]
assert upgraded.ok is True
upgrade = json.loads(upgraded.output)
assert upgrade["generation"] == 2
assert upgrade["previous_sha256"] == first_sha256
second_sha256 = upgrade["definition_sha256"]
assert (
execute_tool_text(
"RunDynamicTool(name='versioned_tool')",
cwd=str(tmp_path),
timeout_s=5,
)[0].output
== "two"
)
retired = execute_tool_text(
f"RetireTool(name='versioned_tool', expected_sha256='{second_sha256}')",
cwd=str(tmp_path),
timeout_s=5,
)[0]
assert retired.ok is True
retirement = json.loads(retired.output)
assert retirement["generation"] == 3
retired_sha256 = retirement["definition_sha256"]
unavailable = execute_tool_text(
"RunDynamicTool(name='versioned_tool')",
cwd=str(tmp_path),
timeout_s=5,
)[0]
assert unavailable.ok is False
assert "retired" in unavailable.error
retired_description = execute_tool_text(
"ToolDescribe(name='versioned_tool')",
cwd=str(tmp_path),
timeout_s=5,
)[0]
assert json.loads(retired_description.output)["status"] == "retired"
restored = execute_tool_text(
"UpgradeTool(name='versioned_tool', command='printf three', "
f"expected_sha256='{retired_sha256}')",
cwd=str(tmp_path),
timeout_s=5,
)[0]
assert restored.ok is True
assert json.loads(restored.output)["generation"] == 4
assert (
execute_tool_text(
"RunDynamicTool(name='versioned_tool')",
cwd=str(tmp_path),
timeout_s=5,
)[0].output
== "three"
)
history = sorted(
(tmp_path / ".nexum" / "tools" / "history" / "versioned_tool").glob("*.json")
)
assert [path.stem for path in history] == [
"000001",
"000002",
"000003",
"000004",
]
current = tmp_path / ".nexum" / "tools" / "current" / "versioned_tool.json"
altered = json.loads(current.read_text(encoding="utf-8"))
altered["command"] = "printf altered"
current.write_text(json.dumps(altered), encoding="utf-8")
tampered = execute_tool_text(
"RunDynamicTool(name='versioned_tool')",
cwd=str(tmp_path),
timeout_s=5,
)[0]
assert tampered.ok is False
assert "integrity check failed" in tampered.error
def test_multi_tool_parser_preserves_apostrophes_and_order(tmp_path: Path) -> None:
text = (
f'{TOOL_CALL_START}[Write(path="note.txt", content="can\'t fail"), '
f'Read(path="note.txt")]{TOOL_CALL_END}'
)
calls = parse_tool_calls(text)
assert [(call.name, call.args) for call in calls] == [
("Write", {"path": "note.txt", "content": "can't fail"}),
("Read", {"path": "note.txt"}),
]
results = execute_tool_text(text, cwd=str(tmp_path), timeout_s=5)
assert [result.ok for result in results] == [True, True]
assert results[1].output == "can't fail"
def test_tool_parser_accepts_equivalent_json_function_protocols() -> None:
text = (
f"{TOOL_CALL_START}"
'[{"id":"call_write","type":"function","function":{"name":"Write",'
'"arguments":"{\\"path\\":\\"note.txt\\",\\"content\\":\\"ready\\"}"}},'
'{"name":"Read","arguments":{"path":"note.txt",'
'"depends_on":["call_write"]}}]'
f"{TOOL_CALL_END}"
)
calls = parse_tool_calls(text)
assert [(call.name, call.args) for call in calls] == [
("Write", {"path": "note.txt", "content": "ready"}),
("Read", {"path": "note.txt"}),
]
assert calls[0].call_id == "call_write"
assert calls[1].depends_on == ("call_write",)
def test_tool_parser_accepts_one_spurious_closer_without_rewriting_action() -> None:
text = (
f"{TOOL_CALL_START}"
"[Inspect(payload={'reference': 'current-71'}, modes=['verified']])]"
f"{TOOL_CALL_END}"
)
calls = parse_tool_calls(text)
assert [(call.name, call.args) for call in calls] == [
(
"Inspect",
{
"payload": {"reference": "current-71"},
"modes": ["verified"],
},
)
]
assert calls[0].raw == text
def test_tool_parser_classifies_unterminated_model_action_as_invalid() -> None:
marked = f"{TOOL_CALL_START}[Bash(command='python inspect_state.py]{TOOL_CALL_END}"
diagnostic = inspect_tool_text(marked, {"Bash"})
assert diagnostic["valid"] is False
assert diagnostic["reason"] == "invalid_tool_syntax"
assert diagnostic["tool_count"] == 0
assert parse_tool_calls("Bash(command='python inspect_state.py") == []
boundary = _tool_protocol_boundary(marked, {"Bash"})
assert boundary.action_detected is True
assert boundary.valid is False
assert boundary.reason == "invalid_tool_syntax"
assert not boundary.executable
def test_runtime_tool_schemas_accept_dependency_metadata() -> None:
for schema in tool_schemas():
function = schema["function"]
parameters = function["parameters"]
assert parameters["additionalProperties"] is False
assert "depends_on" in parameters["properties"]
assert "depends_on" not in parameters["required"]
def test_model_selected_dependencies_order_runtime_tools(tmp_path: Path) -> None:
calls = (
ToolCall(
name="Write",
args={"path": "ordered.txt", "content": "ready"},
raw="",
call_id="call_write",
),
ToolCall(
name="Read",
args={"path": "ordered.txt", "depends_on": ["call_write"]},
raw="",
call_id="call_read",
depends_on=("call_write",),
),
)
results = execute_tool_calls(calls, cwd=str(tmp_path), timeout_s=5)
assert [result.ok for result in results] == [True, True]
assert results[1].output == "ready"
assert results[1].args == {"path": "ordered.txt"}
def test_marked_tool_call_dependency_metadata_is_not_an_argument() -> None:
call = parse_tool_calls("Read(path='ordered.txt', depends_on=['call_write'])")[0]
assert call.depends_on == ("call_write",)
assert call.args == {"path": "ordered.txt"}
def test_file_and_search_tools_cannot_leave_workspace(tmp_path: Path) -> None:
outside = tmp_path.parent / "outside.txt"
outside.write_text("private", encoding="utf-8")
read = execute_tool_text(
f"{TOOL_CALL_START}[Read(path='../outside.txt')]{TOOL_CALL_END}",
cwd=str(tmp_path),
)[0]
glob = execute_tool_text(
f"{TOOL_CALL_START}[Glob(pattern='../*.txt')]{TOOL_CALL_END}",
cwd=str(tmp_path),
)[0]
grep = execute_tool_text(
f"{TOOL_CALL_START}[Grep(pattern='private', path='../')]{TOOL_CALL_END}",
cwd=str(tmp_path),
)[0]
assert read.ok is False
assert "use a workspace-relative path" in read.error
assert glob.ok is False
assert grep.ok is False
@pytest.mark.parametrize(
"url",
(
"file:///etc/passwd",
"http://127.0.0.1/private",
"http://[::1]/private",
"http://169.254.169.254/latest/meta-data/",
"http://user:password@example.com/",
),
)
def test_web_fetch_rejects_non_public_targets(tmp_path: Path, url: str) -> None:
result = execute_tool_text(
f"WebFetch(url={url!r})", cwd=str(tmp_path), timeout_s=1
)[0]
assert result.ok is False
assert result.executed is True
def test_web_fetch_connects_to_the_address_that_was_validated() -> None:
captured: dict[str, object] = {}
class FakeSocket:
def getpeername(self) -> tuple[str, int]:
return "93.184.216.34", 80
class FakeResponse:
status = 200
class FakeConnection:
sock = FakeSocket()
def __init__(self, host: str, *, port: int, timeout: float) -> None:
captured.update(host=host, port=port, timeout=timeout)
def request(self, method: str, path: str, *, headers: dict[str, str]) -> None:
captured.update(method=method, path=path, headers=headers)
def getresponse(self) -> FakeResponse:
return FakeResponse()
def close(self) -> None:
return
parsed = urllib.parse.urlsplit("http://example.com/path?q=1")
with mock.patch(
"nexum_runtime.executor.http.client.HTTPConnection", FakeConnection
):
connection, _response = _public_http_response(parsed, ("93.184.216.34",), 3.0)
connection.close()
assert captured["host"] == "93.184.216.34"
assert captured["headers"] == {
"Host": "example.com",
"User-Agent": "nexum-runtime/0.1",
}
def test_model_driven_harness_corrects_after_real_failure(tmp_path: Path) -> None:
payloads: list[dict[str, object]] = []
def complete(payload: dict[str, object]) -> dict[str, object]:
payloads.append(payload)
if len(payloads) == 1:
return _completion(
{
"role": "assistant",
"content": None,
"tool_calls": [
_tool_call("call_bad", "Bash", {"command": "false"})
],
}
)
if len(payloads) == 2:
return _completion(
{
"role": "assistant",
"content": None,
"tool_calls": [
_tool_call("call_fixed", "Bash", {"command": "printf repaired"})
],
}
)
return _completion(
{"role": "assistant", "content": "completed from observed evidence"}
)
response = NNFXHarnessSession(
complete=complete,
sign_observation=lambda _session_id, row: {
**row,
"receipt_nonce": "test-nonce",
"receipt_auth": "test-auth",
},
).run(NNFXHarnessRequest(prompt="complete the task", workspace=str(tmp_path)))
assert response.ok is True
assert response.final_text == "completed from observed evidence"
assert [row["ok"] for row in response.tool_results] == [False, True]
assert len(payloads) == 3
initial_messages = cast(list[dict[str, object]], payloads[0]["messages"])
assert initial_messages[0]["role"] == "system"
assert "use accumulated observations" in cast(str, initial_messages[0]["content"])
second_observations = cast(
list[dict[str, object]], payloads[1]["nexum_observations"]
)
third_observations = cast(
list[dict[str, object]], payloads[2]["nexum_observations"]
)
assert second_observations[0]["executed"] is True
assert third_observations[0]["output"] == "repaired"
assert len(response.receipt_ids) == 2
def test_harness_resumes_private_open_trajectory_after_completion_crash(
tmp_path: Path,
) -> None:
session_id = "private-resume"
calls = 0
def interrupted(payload: dict[str, object]) -> dict[str, object]:
nonlocal calls
calls += 1
if calls == 1:
return _completion(
{
"role": "assistant",
"content": None,
"tool_calls": [
_tool_call("call_resume", "Bash", {"command": "printf durable"})
],
}
)
observations = cast(list[dict[str, object]], payload["nexum_observations"])
assert observations[0]["output"] == "durable"
raise RuntimeError("simulated generation crash")
def signer(_session_id: str, row: dict[str, object]) -> dict[str, object]:
return {
**row,
"receipt_nonce": "durable-nonce",
"receipt_auth": "durable-auth",
}
request = NNFXHarnessRequest(
prompt="complete the durable task",
workspace=str(tmp_path),
session_id=session_id,
)
with pytest.raises(RuntimeError, match="simulated generation crash"):
NNFXHarnessSession(
complete=interrupted,
sign_observation=signer,
).run(request)
state_path = next(
(tmp_path / ".nexum" / "private" / "sessions").glob("*/state.json")
)
assert state_path.stat().st_mode & 0o777 == 0o600
def resumed(payload: dict[str, object]) -> dict[str, object]:
observations = cast(list[dict[str, object]], payload["nexum_observations"])
assert observations[0]["receipt_nonce"] == "durable-nonce"
messages = cast(list[dict[str, object]], payload["messages"])
assert any(
row.get("role") == "tool" and "durable" in str(row.get("content"))
for row in messages
)
return _completion({"role": "assistant", "content": "resumed"})
response = NNFXHarnessSession(complete=resumed).run(request)
assert response.final_text == "resumed"
assert not state_path.exists()
def test_harness_resumes_exact_caller_owned_tool_result_as_self_correction(
tmp_path: Path,
) -> None:
session_id = "external-correction"
external_schema = {
"type": "function",
"function": {
"name": "CallerLookup",
"description": "Look up caller-owned state.",
"parameters": {
"type": "object",
"properties": {"key": {"type": "string"}},
"required": ["key"],
"additionalProperties": False,
},
},
}
first = NNFXHarnessSession(
complete=lambda _payload: _completion(
{
"role": "assistant",
"content": None,
"tool_calls": [
_tool_call("call_external", "CallerLookup", {"key": "release"})
],
}
)
).run(
NNFXHarnessRequest(
prompt="inspect caller state",
workspace=str(tmp_path),
session_id=session_id,
tools=(external_schema,),
)
)
assert first.message == "external_tool_required"
assert first.tool_calls[0]["id"] == "call_external"
observed_payloads: list[dict[str, object]] = []
def resumed(payload: dict[str, object]) -> dict[str, object]:
observed_payloads.append(payload)
observations = cast(list[dict[str, object]], payload["nexum_observations"])
assert observations[0]["receipt_auth"] == "external-auth"
assert observations[0]["source_trust"] == "caller_owned"
messages = cast(list[dict[str, object]], payload["messages"])
assert any(
row.get("role") == "tool" and "release-ready" in str(row.get("content"))
for row in messages
)
return _completion({"role": "assistant", "content": "verified"})
corrected = NNFXHarnessSession(
complete=resumed,
sign_observation=lambda _session_id, row: {
**row,
"receipt_nonce": "external-nonce",
"receipt_auth": "external-auth",
},
).run(
NNFXHarnessRequest(
prompt="inspect caller state",
workspace=str(tmp_path),
session_id=session_id,
tools=(external_schema,),
tool_results=(
{
"tool_call_id": "call_external",
"name": "CallerLookup",
"args": {"key": "release"},
"ok": True,
"executed": True,
"output": "release-ready",
},
),
)
)
assert corrected.final_text == "verified"
assert len(observed_payloads) == 1
assert corrected.tool_results[0]["source_trust"] == "caller_owned"
def test_model_can_fan_out_caller_agents_and_resume_partial_results(
tmp_path: Path,
) -> None:
session_id = "parallel-agent-handoff"
selected_calls = [
_tool_call(
f"call_agent_{index}",
"AgentDelegate",
{
"agent": f"agent-{index}",
"objective": f"Produce independently verifiable evidence {index}",
},
)
for index in range(3)
]
def select_agents(payload: dict[str, object]) -> dict[str, object]:
schemas = cast(list[dict[str, object]], payload["tools"])
names = {
cast(dict[str, object], schema["function"])["name"] for schema in schemas
}
assert {"AgentDiscover", "AgentDelegate", "AgentTaskStatus"} <= names
return _completion(
{
"role": "assistant",
"content": None,
"tool_calls": selected_calls,
}
)
request = NNFXHarnessRequest(
prompt="Inspect three independent surfaces concurrently",
workspace=str(tmp_path),
session_id=session_id,
)
first = NNFXHarnessSession(complete=select_agents).run(request)
assert first.message == "external_tool_required"
assert [call["id"] for call in first.tool_calls] == [
"call_agent_0",
"call_agent_1",
"call_agent_2",
]
assert first.tool_results == ()
def signer(_session_id: str, row: dict[str, object]) -> dict[str, object]:
return {
**row,
"receipt_nonce": "agent-nonce",
"receipt_auth": "agent-auth",
}
partial = NNFXHarnessSession(
complete=lambda _payload: pytest.fail(
"generation resumed before fan-out returned"
),
sign_observation=signer,
).run(
NNFXHarnessRequest(
prompt=request.prompt,
workspace=request.workspace,
session_id=session_id,
tool_results=(
{
"tool_call_id": "call_agent_1",
"name": "AgentDelegate",
"args": {
"agent": "agent-1",
"objective": "Produce independently verifiable evidence 1",
},
"ok": True,
"executed": True,
"output": "evidence-1",
},
),
)
)
assert [call["id"] for call in partial.tool_calls] == [
"call_agent_0",
"call_agent_2",
]
observed_payloads: list[dict[str, object]] = []
def finish(payload: dict[str, object]) -> dict[str, object]:
observed_payloads.append(payload)
observations = cast(list[dict[str, object]], payload["nexum_observations"])
assert len(observations) == 3
assert {row["receipt_source"] for row in observations} == {"caller_attested"}
assert {row["source_trust"] for row in observations} == {"caller_owned"}
return _completion(
{"role": "assistant", "content": "combined verified evidence"}
)
final = NNFXHarnessSession(
complete=finish,
sign_observation=signer,
).run(
NNFXHarnessRequest(
prompt=request.prompt,
workspace=request.workspace,
session_id=session_id,
tool_results=tuple(
{
"tool_call_id": f"call_agent_{index}",
"name": "AgentDelegate",
"args": {
"agent": f"agent-{index}",
"objective": f"Produce independently verifiable evidence {index}",
},
"ok": True,
"executed": True,
"output": f"evidence-{index}",
}
for index in (0, 2)
),
)
)
assert final.final_text == "combined verified evidence"
assert len(final.tool_results) == 3
assert len(observed_payloads) == 1
def test_caller_agent_results_honor_model_selected_dependencies(
tmp_path: Path,
) -> None:
parent = _tool_call(
"call_parent",
"AgentDelegate",
{"agent": "research", "objective": "Collect evidence"},
)
child = _tool_call(
"call_child",
"AgentDelegate",
{
"agent": "review",
"objective": "Review collected evidence",
"depends_on": ["call_parent"],
},
)
request = NNFXHarnessRequest(
prompt="Collect and review evidence",
workspace=str(tmp_path),
session_id="dependent-agent-handoff",
)
first = NNFXHarnessSession(
complete=lambda _payload: _completion(
{
"role": "assistant",
"content": None,
"tool_calls": [parent, child],
}
)
).run(request)
assert len(first.tool_calls) == 2
child_result = {
"tool_call_id": "call_child",
"name": "AgentDelegate",
"args": {"agent": "review", "objective": "Review collected evidence"},
"ok": True,
"executed": True,
"output": "reviewed",
}
with pytest.raises(ValueError, match="unresolved dependencies"):
NNFXHarnessSession(
complete=lambda _payload: pytest.fail("invalid dependency resumed"),
sign_observation=lambda _session_id, row: row,
).run(
NNFXHarnessRequest(
prompt=request.prompt,
workspace=request.workspace,
session_id=request.session_id,
tool_results=(child_result,),
)
)
observed_payloads: list[dict[str, object]] = []
def finish(payload: dict[str, object]) -> dict[str, object]:
observed_payloads.append(payload)
observations = cast(list[dict[str, object]], payload["nexum_observations"])
assert [row["tool_call_id"] for row in observations] == [
"call_parent",
"call_child",
]
return _completion({"role": "assistant", "content": "dependency honored"})
final = NNFXHarnessSession(
complete=finish,
sign_observation=lambda _session_id, row: row,
).run(
NNFXHarnessRequest(
prompt=request.prompt,
workspace=request.workspace,
session_id=request.session_id,
tool_results=(
{
"tool_call_id": "call_parent",
"name": "AgentDelegate",
"args": {"agent": "research", "objective": "Collect evidence"},
"ok": True,
"executed": True,
"output": "collected",
},
child_result,
),
)
)
assert final.final_text == "dependency honored"
assert len(observed_payloads) == 1
def test_harness_rejects_unbound_or_evaluator_shaped_external_results(
tmp_path: Path,
) -> None:
session_id = "external-result-rejection"
external_schema = {
"type": "function",
"function": {
"name": "CallerLookup",
"description": "Look up caller-owned state.",
"parameters": {
"type": "object",
"properties": {"key": {"type": "string"}},
"required": ["key"],
"additionalProperties": False,
},
},
}
request = NNFXHarnessRequest(
prompt="inspect caller state",
workspace=str(tmp_path),
session_id=session_id,
tools=(external_schema,),
)
first = NNFXHarnessSession(
complete=lambda _payload: _completion(
{
"role": "assistant",
"content": None,
"tool_calls": [
_tool_call("call_exact", "CallerLookup", {"key": "release"})
],
}
)
).run(request)
assert first.message == "external_tool_required"
for invalid in (
{
"tool_call_id": "call_wrong",
"name": "CallerLookup",
"args": {"key": "release"},
"ok": True,
"executed": True,
"output": "value",
},
{
"tool_call_id": "call_exact",
"name": "CallerLookup",
"args": {"key": "different"},
"ok": True,
"executed": True,
"output": "value",
},
{
"tool_call_id": "call_exact",
"name": "CallerLookup",
"args": {"key": "release"},
"ok": True,
"executed": True,
"output": "value",
"score": 1.0,
},
):
with pytest.raises(ValueError):
NNFXHarnessSession(
complete=lambda _payload: _completion(
{"role": "assistant", "content": "unreachable"}
),
sign_observation=lambda _session_id, row: row,
).run(
NNFXHarnessRequest(
**{
**asdict(request),
"tool_results": (invalid,),
}
)
)
def test_harness_resumes_schema_validated_structured_input(tmp_path: Path) -> None:
session_id = "structured-input"
first = NNFXHarnessSession(
complete=lambda _payload: _completion(
{
"role": "assistant",
"content": None,
"tool_calls": [
_tool_call(
"call-input",
"RequestInput",
{
"prompt": "Choose a region",
"schema": {
"type": "string",
"enum": ["east", "west"],
},
},
)
],
}
)
).run(
NNFXHarnessRequest(
prompt="configure the region",
workspace=str(tmp_path),
session_id=session_id,
)
)
assert first.message == "input_required"
result_row = {
"tool_call_id": "call-input",
"name": "RequestInput",
"args": {
"prompt": "Choose a region",
"schema": {"type": "string", "enum": ["east", "west"]},
},
"ok": True,
"executed": True,
"output": json.dumps("west"),
}
with pytest.raises(ValueError, match="does not satisfy"):
NNFXHarnessSession(
complete=lambda _payload: _completion(
{"role": "assistant", "content": "unreachable"}
),
sign_observation=lambda _session_id, row: row,
).run(
NNFXHarnessRequest(
prompt="configure the region",
workspace=str(tmp_path),
session_id=session_id,
tool_results=({**result_row, "output": json.dumps("north")},),
)
)
def complete(payload: dict[str, object]) -> dict[str, object]:
observations = cast(list[dict[str, object]], payload["nexum_observations"])
assert observations[0]["output"] == json.dumps("west")
return _completion({"role": "assistant", "content": "region configured"})
response = NNFXHarnessSession(
complete=complete,
sign_observation=lambda _session_id, row: {
**row,
"receipt_nonce": "input-nonce",
"receipt_auth": "input-auth",
},
).run(
NNFXHarnessRequest(
prompt="configure the region",
workspace=str(tmp_path),
session_id=session_id,
tool_results=(result_row,),
)
)
assert response.final_text == "region configured"
def test_harness_honors_external_cancellation_without_a_step_cap(
tmp_path: Path,
) -> None:
calls = 0
def complete(_payload: dict[str, object]) -> dict[str, object]:
nonlocal calls
calls += 1
return _completion(
{
"role": "assistant",
"content": None,
"tool_calls": [_tool_call("call_once", "Bash", {"command": "true"})],
}
)
response = NNFXHarnessSession(
complete=complete,
cancelled=lambda: calls > 0,
).run(NNFXHarnessRequest(prompt="continue", workspace=str(tmp_path)))
assert response.ok is False
assert response.message == "client_disconnected"
assert calls == 1
def test_harness_executes_only_tools_advertised_by_request(tmp_path: Path) -> None:
calls = 0
payloads: list[dict[str, object]] = []
def complete(payload: dict[str, object]) -> dict[str, object]:
nonlocal calls
calls += 1
payloads.append(payload)
if calls == 1:
return _completion(
{
"role": "assistant",
"content": None,
"tool_calls": [
_tool_call(
"call_hidden",
"Bash",
{"command": "printf unsafe > should-not-exist"},
)
],
}
)
return _completion({"role": "assistant", "content": "rejected"})
external_schema = {
"type": "function",
"function": {
"name": "ExternalSafe",
"description": "Caller-owned tool",
"parameters": {"type": "object", "properties": {}},
},
}
response = NNFXHarnessSession(
complete=complete,
sign_observation=lambda _session_id, row: {
**row,
"receipt_nonce": "rejected-nonce",
"receipt_auth": "rejected-auth",
},
).run(
NNFXHarnessRequest(
prompt="continue", workspace=str(tmp_path), tools=(external_schema,)
)
)
assert response.ok is True
assert response.tool_results[0]["executed"] is False
assert response.tool_results[0]["tool_call_id"] == "call_hidden"
assert len(response.receipt_ids) == 1
observations = cast(list[dict[str, object]], payloads[1]["nexum_observations"])
assert observations[0]["tool_call_id"] == "call_hidden"
assert not (tmp_path / "should-not-exist").exists()
def test_harness_preserves_repeated_model_selected_actions(tmp_path: Path) -> None:
calls = 0
def complete(_payload: dict[str, object]) -> dict[str, object]:
nonlocal calls
calls += 1
if calls <= 2:
return _completion(
{
"role": "assistant",
"content": None,
"tool_calls": [
_tool_call(
f"call_{calls}",
"Bash",
{"command": "printf x >> repeated.txt"},
)
],
}
)
return _completion({"role": "assistant", "content": "done"})
response = NNFXHarnessSession(complete=complete).run(
NNFXHarnessRequest(prompt="repeat exactly", workspace=str(tmp_path))
)
assert [row["executed"] for row in response.tool_results] == [True, True]
assert (tmp_path / "repeated.txt").read_text(encoding="utf-8") == "xx"
def test_harness_returns_model_turn_and_repairs_malformed_tool_call(
tmp_path: Path,
) -> None:
payloads: list[dict[str, object]] = []
def complete(payload: dict[str, object]) -> dict[str, object]:
payloads.append(payload)
if len(payloads) == 1:
response = _completion(
{
"role": "assistant",
"content": None,
"tool_calls": [
{
"id": "call-invalid",
"type": "function",
"function": {"name": "Bash", "arguments": "{"},
}
],
}
)
response["nexum"] = {"session_turn": 4}
return response
response = _completion({"role": "assistant", "content": "corrected"})
response["nexum"] = {"session_turn": 5}
return response
response = NNFXHarnessSession(
complete=complete,
sign_observation=lambda _session_id, row: {
**row,
"receipt_nonce": "rejected-nonce",
"receipt_auth": "rejected-auth",
},
).run(NNFXHarnessRequest(prompt="complete", workspace=str(tmp_path)))
assert response.ok is True
assert response.final_text == "corrected"
assert response.metadata["session_turn"] == 5
assert response.tool_results[0]["executed"] is False
assert response.tool_results[0]["tool_call_id"] == "call-invalid"
assert "invalid tool call" in str(response.tool_results[0]["error"])
assert len(response.receipt_ids) == 1
second_messages = payloads[1]["messages"]
assert isinstance(second_messages, list)
assert second_messages[-1]["role"] == "tool"
observations = cast(list[dict[str, object]], payloads[1]["nexum_observations"])
assert observations[0]["tool_call_id"] == "call-invalid"
def test_harness_persists_unparseable_model_action_before_correction(
tmp_path: Path,
) -> None:
payloads: list[dict[str, object]] = []
def complete(payload: dict[str, object]) -> dict[str, object]:
payloads.append(payload)
if len(payloads) == 1:
return _completion(
{
"role": "assistant",
"content": "<|tool_call_start|>Bash(command='printf incomplete')",
}
)
return _completion({"role": "assistant", "content": "corrected"})
response = NNFXHarnessSession(
complete=complete,
sign_observation=lambda _session_id, row: {
**row,
"receipt_nonce": "protocol-nonce",
"receipt_auth": "protocol-auth",
},
).run(NNFXHarnessRequest(prompt="complete", workspace=str(tmp_path)))
assert response.ok is True
assert response.final_text == "corrected"
assert len(response.receipt_ids) == 1
assert response.tool_results[0]["executed"] is False
assert "not parseable" in response.tool_results[0]["error"]
assert "nexum_observations" not in payloads[1]
retry_messages = payloads[1]["messages"]
assert isinstance(retry_messages, list)
assert "not parseable" in str(retry_messages[-1]["content"])
def test_harness_preserves_private_reasoning_and_keeps_objective_open(
tmp_path: Path,
) -> None:
payloads: list[dict[str, object]] = []
(tmp_path / "evidence.txt").write_text("verified", encoding="utf-8")
def complete(payload: dict[str, object]) -> dict[str, object]:
payloads.append(payload)
if len(payloads) == 1:
return _completion({"role": "assistant", "content": ""})
if len(payloads) == 2:
return _completion(
{
"role": "assistant",
"content": None,
"tool_calls": [
_tool_call("call_read", "Read", {"path": "evidence.txt"})
],
}
)
completion = _completion(
{
"role": "assistant",
"content": "preserved private traceverified",
}
)
completion["nexum"] = {
"full_model_active": True,
"tensor_packages_loaded": 113,
"model_authority_engaged": True,
"self_correction_engaged": True,
"recursive_graph_engaged": True,
"recursive_steps": 7,
"recursive_forward_passes": 7,
"recursive_feedback_passes": 7,
"recursive_mutation_passes": 7,
"recursive_reinforcement_passes": 7,
"adaptive_expert_traversal_engaged": True,
"internal_multi_agent_engaged": True,
"internal_agent_banks": 7,
"internal_agent_arms": 12,
"internal_agent_active_arms": 12,
"internal_expert_routes": 84,
"internal_agent_growth_topology_owned": True,
"internal_agent_knowledge_transfer_engaged": True,
"internal_agent_drafting_engaged": True,
"internal_agent_rehearsal_engaged": True,
"internal_agent_challenge_mean": 0.25,
"internal_agent_disagreement_mean": 0.2,
"internal_agent_rehearsal_delta_norm": 0.1,
"internal_agent_release_confidence_mean": 0.8,
"internal_agent_correction_pressure_mean": 0.0,
"model_internal_agents_engaged": True,
"model_internal_agent_count": 12,
"model_internal_agent_contribution_count": 12,
"model_internal_authority_count": 84,
"model_internal_bank_count": 7,
"model_internal_worker_count": 84,
"model_internal_worker_contribution_count": 84,
"model_internal_work_roles_engaged": True,
"model_internal_work_role_coordination_applied": True,
"model_internal_work_role_transfer_applied": True,
"model_internal_drafting_engaged": True,
"model_pre_submission_draft_experimented": True,
"model_pre_submission_draft_experiment_count": 7,
"model_pre_submission_draft_selection_owner": "trained_rbo_confidence",
"model_pre_submission_draft_attempt_cap_active": False,
"model_internal_work_role_count": 5,
"model_internal_work_role_contribution_count": 5,
"model_internal_agent_cap_active": False,
"model_internal_worker_cap_active": False,
"model_internal_bank_cap_active": False,
"model_internal_expert_cap_active": False,
"model_internal_host_fanout": False,
"traversal_layers": 7,
"routed_authority_count": 84,
"model_rotation_applied": True,
"private_internal_name": "not-public",
}
return completion
response = NNFXHarnessSession(complete=complete).run(
NNFXHarnessRequest(prompt="read the evidence", workspace=str(tmp_path))
)
assert response.ok is True
assert response.final_text == "verified"
assert [row["name"] for row in response.tool_results] == ["Read"]
assert len(response.receipt_ids) == 1
assert response.messages[-1]["content"] == (
"preserved private traceverified"
)
assert response.metadata["runtime"] == {
"adaptive_expert_traversal_engaged": True,
"full_model_active": True,
"internal_agent_active_arms": 12,
"internal_agent_arms": 12,
"internal_agent_banks": 7,
"internal_agent_growth_topology_owned": True,
"internal_agent_knowledge_transfer_engaged": True,
"internal_agent_drafting_engaged": True,
"internal_agent_rehearsal_engaged": True,
"internal_agent_challenge_mean": 0.25,
"internal_agent_disagreement_mean": 0.2,
"internal_agent_rehearsal_delta_norm": 0.1,
"internal_agent_release_confidence_mean": 0.8,
"internal_agent_correction_pressure_mean": 0.0,
"internal_expert_routes": 84,
"internal_multi_agent_engaged": True,
"model_internal_agents_engaged": True,
"model_internal_agent_count": 12,
"model_internal_agent_contribution_count": 12,
"model_internal_authority_count": 84,
"model_internal_bank_count": 7,
"model_internal_worker_count": 84,
"model_internal_worker_contribution_count": 84,
"model_internal_work_roles_engaged": True,
"model_internal_work_role_coordination_applied": True,
"model_internal_work_role_transfer_applied": True,
"model_internal_drafting_engaged": True,
"model_pre_submission_draft_experimented": True,
"model_pre_submission_draft_experiment_count": 7,
"model_pre_submission_draft_selection_owner": "trained_rbo_confidence",
"model_pre_submission_draft_attempt_cap_active": False,
"model_internal_work_role_count": 5,
"model_internal_work_role_contribution_count": 5,
"model_internal_agent_cap_active": False,
"model_internal_worker_cap_active": False,
"model_internal_bank_cap_active": False,
"model_internal_expert_cap_active": False,
"model_internal_host_fanout": False,
"model_authority_engaged": True,
"self_correction_engaged": True,
"recursive_graph_engaged": True,
"recursive_steps": 7,
"recursive_forward_passes": 7,
"recursive_feedback_passes": 7,
"recursive_mutation_passes": 7,
"recursive_reinforcement_passes": 7,
"traversal_layers": 7,
"routed_authority_count": 84,
"model_rotation_applied": True,
"tensor_packages_loaded": 113,
}
assert response.metadata["runtime_trajectory"] == {
"adaptive_expert_traversal_engaged_every_forward": True,
"full_model_active_every_forward": True,
"internal_agent_active_arms_max": 12,
"internal_agent_arms_max": 12,
"internal_agent_banks_max": 7,
"internal_agent_growth_topology_owned_every_forward": True,
"internal_agent_knowledge_transfer_engaged_every_forward": True,
"internal_agent_drafting_engaged_every_forward": True,
"internal_agent_rehearsal_engaged_every_forward": True,
"internal_expert_routes_max": 84,
"internal_multi_agent_engaged_every_forward": True,
"model_internal_agents_engaged_every_forward": True,
"model_internal_agent_count_max": 12,
"model_internal_agent_contribution_count_max": 12,
"model_internal_authority_count_max": 84,
"model_internal_bank_count_max": 7,
"model_internal_worker_count_max": 84,
"model_internal_worker_contribution_count_max": 84,
"model_internal_work_roles_engaged_every_forward": True,
"model_internal_work_role_coordination_applied_every_forward": True,
"model_internal_work_role_transfer_applied_every_forward": True,
"model_internal_drafting_engaged_every_forward": True,
"model_internal_work_role_count_max": 5,
"model_internal_work_role_contribution_count_max": 5,
"model_pre_submission_draft_experimented_every_forward": True,
"model_pre_submission_draft_experiment_count_total": 7,
"model_authority_engaged_every_forward": True,
"model_forwards": 1,
"model_rotation_observed": True,
"recursive_feedback_passes_total": 7,
"recursive_forward_passes_total": 7,
"recursive_graph_engaged_every_forward": True,
"recursive_mutation_passes_total": 7,
"recursive_reinforcement_passes_total": 7,
"recursive_steps_max": 7,
"recursive_steps_total": 7,
"routed_authority_count_max": 84,
"self_correction_engaged_every_forward": True,
"tensor_packages_loaded": 113,
"traversal_layers_max": 7,
}
second_messages = payloads[1]["messages"]
assert isinstance(second_messages, list)
assert second_messages[-2]["content"] == ""
assert second_messages[-1]["role"] == "user"
assert "continue the same objective" in second_messages[-1]["content"]
def test_bare_valid_model_action_resolves_unique_case_equivalent_name() -> None:
calls = _wire_tool_calls(
"Bash(command='printf safe')",
{"Bash"},
)
assert len(calls) == 1
assert calls[0]["function"] == {
"name": "Bash",
"arguments": '{"command": "printf safe"}',
}
normalized = _wire_tool_calls(
"bash(command='printf normalized')",
{"Bash"},
)
assert normalized[0]["function"] == {
"name": "Bash",
"arguments": '{"command": "printf normalized"}',
}
assert _wire_tool_calls("Bash(command='printf unsafe')", {"Read"}) == []
assert _wire_tool_calls("Basj(command='printf unsafe')", {"Bash"}) == []
assert _wire_tool_calls(
"BASH(command='printf ambiguous')", {"Bash", "bash"}
) == []
assert _wire_tool_calls("Bash(command='printf unsafe')") == []
assert _wire_tool_calls("The task is complete.", {"Bash"}) == []
def test_protocol_boundary_rejects_unknown_names_and_invalid_arguments() -> None:
schema = {
"type": "function",
"function": {
"name": "ExactLookup",
"parameters": {
"type": "object",
"properties": {"record_ref": {"type": "string"}},
"required": ["record_ref"],
"additionalProperties": False,
},
},
}
unknown_text = f"{TOOL_CALL_START}[ExactLookpu(record_ref='stale')]{TOOL_CALL_END}"
unknown = _tool_protocol_boundary(unknown_text, {"ExactLookup"}, [schema])
assert unknown.valid is False
assert unknown.reason == "unknown_tool"
assert unknown.unknown_count == 1
assert unknown.invalid_argument_count == 0
assert not unknown.executable
assert unknown.rejected[0]["function"]["name"] == "ExactLookpu"
normalized_text = (
f"{TOOL_CALL_START}[exactlookup(record_ref='current')]{TOOL_CALL_END}"
)
normalized = _tool_protocol_boundary(
normalized_text, {"ExactLookup"}, [schema]
)
assert normalized.valid is True
assert normalized.reason == "valid"
assert normalized.normalized_count == 1
assert normalized.unknown_count == 0
assert normalized.executable[0]["function"] == {
"name": "ExactLookup",
"arguments": '{"record_ref": "current"}',
}
ambiguous = _tool_protocol_boundary(
f"{TOOL_CALL_START}[EXACTLOOKUP(record_ref='current')]{TOOL_CALL_END}",
{"ExactLookup", "exactlookup"},
[schema],
)
assert ambiguous.valid is False
assert ambiguous.reason == "unknown_tool"
assert ambiguous.normalized_count == 0
invalid_text = f"{TOOL_CALL_START}[ExactLookup(record_reff='stale')]{TOOL_CALL_END}"
invalid = _tool_protocol_boundary(invalid_text, {"ExactLookup"}, [schema])
assert invalid.valid is False
assert invalid.reason == "invalid_tool_arguments"
assert invalid.unknown_count == 0
assert invalid.invalid_argument_count == 1
assert not invalid.executable
assert invalid.rejected[0]["function"]["name"] == "ExactLookup"
normalized_invalid = _tool_protocol_boundary(
f"{TOOL_CALL_START}[exactlookup(record_reff='stale')]{TOOL_CALL_END}",
{"ExactLookup"},
[schema],
)
assert normalized_invalid.valid is False
assert normalized_invalid.reason == "invalid_tool_arguments"
assert normalized_invalid.normalized_count == 1
assert normalized_invalid.invalid_argument_count == 1
assert not normalized_invalid.executable
assert (
inspect_tool_text(
"ExactLookup(record_ref='current')", {"ExactLookup"}, [schema]
)["valid"]
is True
)
def test_protocol_rejection_preserves_action_without_executing_it() -> None:
call = _tool_call(
"call_rejected_exact",
"Bash",
{"command": "printf must-not-run"},
)
schema = {
"type": "function",
"function": {
"name": "ExactLookup",
"description": "private prose is not correction state",
"parameters": {
"type": "object",
"properties": {"record_ref": {"type": "string"}},
"required": ["record_ref"],
},
},
}
result = _protocol_rejection_result(call, "unknown_tool", [schema])
assert result["tool_call_id"] == "call_rejected_exact"
assert result["name"] == "Bash"
assert result["args"] == {"command": "printf must-not-run"}
assert result["ok"] is False
assert result["executed"] is False
assert result["status"] == "rejected"
assert result["output"] == ""
assert result["source_trust"] == "trusted_execution"
assert "exact supplied function name" in result["error"]
assert result["supplied_contracts"] == [
{
"name": "ExactLookup",
"argument_names": ["record_ref"],
"required_argument_names": ["record_ref"],
}
]
assert "private prose" not in json.dumps(result)
def test_nnfx_authenticates_unknown_model_action_before_exact_schema_retry(
tmp_path: Path,
) -> None:
schema = {
"type": "function",
"function": {
"name": "ExactLookup",
"parameters": {
"type": "object",
"properties": {"record_ref": {"type": "string"}},
"required": ["record_ref"],
"additionalProperties": False,
},
},
}
payloads: list[dict[str, object]] = []
def complete(payload: dict[str, object]) -> dict[str, object]:
payloads.append(payload)
if len(payloads) == 1:
response = _completion(
{
"role": "assistant",
"content": (
f"{TOOL_CALL_START}[ExactLookpu(record_ref='stale')]"
f"{TOOL_CALL_END}"
),
}
)
response["nexum"] = {
"tool_action_protocol_detected": True,
"tool_action_protocol_valid": False,
"tool_action_protocol_reason": "unknown_tool",
"tool_action_protocol_tool_count": 1,
"tool_action_protocol_unknown_count": 1,
"tool_action_protocol_invalid_argument_count": 0,
"tool_action_protocol_rejection_ids": ["call_rejected_exact"],
"tool_action_protocol_rejected_calls": [
_tool_call(
"call_rejected_exact",
"ExactLookpu",
{"record_ref": "stale"},
)
],
}
return response
return _completion(
{
"role": "assistant",
"content": None,
"tool_calls": [
_tool_call(
"call_exact",
"ExactLookup",
{"record_ref": "stale"},
)
],
}
)
response = NNFXHarnessSession(
complete=complete,
sign_observation=lambda _session_id, row: {
**row,
"receipt_nonce": "protocol-nonce",
"receipt_auth": "protocol-auth",
},
).run(
NNFXHarnessRequest(
prompt="read the exact record",
workspace=str(tmp_path),
session_id="protocol-correction",
tools=(schema,),
)
)
assert response.message == "external_tool_required"
assert response.tool_calls[0]["function"] == {
"name": "ExactLookup",
"arguments": '{"record_ref": "stale"}',
}
assert response.tool_results[0]["name"] == "ExactLookpu"
assert response.tool_results[0]["executed"] is False
second_observations = cast(
list[dict[str, object]], payloads[1]["nexum_observations"]
)
assert second_observations[0]["tool_call_id"] == "call_rejected_exact"
assert second_observations[0]["receipt_source"] == "runtime_rejection"
second_messages = cast(list[dict[str, object]], payloads[1]["messages"])
assert second_messages[-2]["role"] == "tool"
assert second_messages[-2]["name"] == "ExactLookpu"
assert "exact supplied function name" in str(second_messages[-2]["content"])
assert "ExactLookup" in str(second_messages[-2]["content"])
assert "record_ref" in str(second_messages[-2]["content"])
assert second_messages[-1]["role"] == "user"
assert "exact function and argument names" in str(second_messages[-1]["content"])
def test_chat_template_preserves_opaque_contracts_and_success_completion() -> None:
template = (MODEL_DIR / "chat_template.jinja").read_text(encoding="utf-8")
assert "treat those names as opaque identifiers" in template
assert "copy the selected schema's complete function name" in template
assert "After an authenticated successful tool result" in template
assert "specific unresolved requirement" in template
def test_harness_rejects_unsupported_learning_inputs() -> None:
with pytest.raises(ValueError, match="unsupported NNF X request fields"):
NNFXHarnessRequest.from_json(
{
"prompt": "continue",
"external_learning_signal": {"value": 1.0},
}
)
def test_tool_end_stopping_criterion_stays_tensor_native() -> None:
tokenizer = SimpleNamespace(encode=lambda *_args, **_kwargs: [7, 8])
criteria = _native_tool_call_stopping_criteria(
tokenizer,
prompt_len=2,
tools_available=True,
)
assert criteria is not None
running = criteria(torch.tensor([[1, 2, 7]]), scores=None)
complete = criteria(torch.tensor([[1, 2, 7, 8]]), scores=None)
assert isinstance(running, torch.Tensor)
assert running.dtype == torch.bool
assert torch.equal(running, torch.tensor([False]))
assert torch.equal(complete, torch.tensor([True]))
def test_native_confidence_stopping_criterion_stays_model_owned() -> None:
tokenizer = SimpleNamespace(encode=lambda *_args, **_kwargs: [7, 8])
owner = SimpleNamespace(native_decode_stop_t=lambda: torch.tensor([True]))
criteria = _native_tool_call_stopping_criteria(
tokenizer, prompt_len=2, confidence_owner=owner
)
assert criteria is not None
stopped = criteria(torch.tensor([[1, 2, 6]]), scores=None)
assert torch.equal(stopped, torch.tensor([True]))
def test_native_live_analysis_is_diagnostic_only(
monkeypatch: pytest.MonkeyPatch,
capsys: pytest.CaptureFixture[str],
) -> None:
class Tokenizer:
all_special_ids: list[int] = []
@staticmethod
def encode(_text: str, **_kwargs: object) -> list[int]:
return [7, 8]
@staticmethod
def decode(token_ids: list[int], **_kwargs: object) -> str:
return f"token-{token_ids[0]}"
owner = SimpleNamespace(
native_decode_stop_t=lambda: torch.tensor([True]),
native_decode_analysis_t=lambda: torch.tensor(
[[0.9, 0.1, 0.5, 2.2, 3.0, 1.0, 0.0, 0.0, 0.0, 0.25]]
),
)
monkeypatch.setenv("NEXUM_LIVE_ANALYSIS", "1")
criteria = _native_tool_call_stopping_criteria(
Tokenizer(), prompt_len=2, confidence_owner=owner
)
assert criteria is not None
stopped = criteria(torch.tensor([[1, 2, 11]]), scores=None)
event = json.loads(capsys.readouterr().out)
assert torch.equal(stopped, torch.tensor([True]))
assert event["event"] == "nexum_live_decode"
assert event["generated_tokens"] == 1
assert event["token_id"] == 11
assert event["token_text"] == "token-11"
assert event["model_stop_candidate"] is True
assert event["criterion_stop"] is True
assert event["recursive_generation_forwards"] == 3
assert event["trajectory_tokens"] == 1
assert event["completion_context_delta_norm"] == 0.25
def test_native_confidence_cannot_interrupt_an_open_tool_frame() -> None:
class Tokenizer:
@staticmethod
def encode(text: str, **_kwargs: object) -> list[int]:
return [5, 6] if text == TOOL_CALL_START else [7, 8]
owner = SimpleNamespace(native_decode_stop_t=lambda: torch.tensor([True]))
criteria = _native_tool_call_stopping_criteria(
Tokenizer(),
prompt_len=2,
confidence_owner=owner,
tools_available=True,
)
assert criteria is not None
partial_marker = criteria(torch.tensor([[1, 2, 5]]), scores=None)
open_frame = criteria(torch.tensor([[1, 2, 5, 6, 11]]), scores=None)
closed_frame = criteria(torch.tensor([[1, 2, 5, 6, 11, 7, 8]]), scores=None)
assert torch.equal(partial_marker, torch.tensor([False]))
assert torch.equal(open_frame, torch.tensor([False]))
assert torch.equal(closed_frame, torch.tensor([True]))
def test_native_confidence_requires_a_public_emission() -> None:
class Tokenizer:
all_special_ids = [9]
@staticmethod
def encode(text: str, **_kwargs: object) -> list[int]:
return [5, 6] if text == TOOL_CALL_START else [7, 8]
owner = SimpleNamespace(native_decode_stop_t=lambda: torch.tensor([True]))
criteria = _native_tool_call_stopping_criteria(
Tokenizer(),
prompt_len=2,
confidence_owner=owner,
)
assert criteria is not None
special_only = criteria(torch.tensor([[1, 2, 9]]), scores=None)
visible = criteria(torch.tensor([[1, 2, 9, 11]]), scores=None)
assert torch.equal(special_only, torch.tensor([False]))
assert torch.equal(visible, torch.tensor([True]))
def test_native_confidence_readiness_rejects_initialization_values() -> None:
surface = NativeDecodeConfidenceSurface(NexumConfig.tiny())
model = SimpleNamespace(lm_head=SimpleNamespace(native_decode_confidence=surface))
assert _native_confidence_checkpoint_ready(model) is False
with torch.no_grad():
surface.complete_head.bias.add_(0.25)
assert _native_confidence_checkpoint_ready(model) is True
def test_native_confidence_training_balances_completion_and_fragments() -> None:
cfg = NexumConfig.tiny()
surface = NativeDecodeConfidenceSurface(cfg)
hidden = torch.randn(1, 5, cfg.hidden_size)
logits = torch.randn(1, 5, cfg.vocab_size)
labels = torch.tensor([[-100, 11, 12, 13, 14]])
verified_completion_t = torch.tensor([[False, False, True, True]])
loss, receipt = native_decode_confidence_training_loss(
surface,
hidden,
logits,
labels,
verified_completion_t=verified_completion_t,
)
torch.autograd.backward(loss)
assert torch.isfinite(loss)
assert receipt["native_decode_confidence_positive"] == 1
assert receipt["native_decode_confidence_negative"] == 2
assert receipt["native_decode_confidence_ignored_after_verified"] == 1
assert receipt["native_decode_confidence_tokens_to_verified_mean"] == 3.0
assert receipt["native_decode_confidence_hard_negative_loss"] > 0
assert surface.complete_head.bias.grad is not None
assert surface.complete_head.bias.grad.item() < 0
assert surface.fragment_head.bias.grad is not None
assert surface.fragment_head.bias.grad.item() > 0
def test_native_confidence_cached_features_match_logits_path() -> None:
surface = NativeDecodeConfidenceSurface(NexumConfig.tiny())
hidden = torch.randn(2, 3, NexumConfig.tiny().hidden_size)
logits = torch.randn(2, 3, 17)
features = surface._logit_features(logits)
probabilities = torch.softmax(logits.float(), dim=-1)
top = torch.topk(probabilities, k=2, dim=-1).values
entropy = -(
probabilities * probabilities.clamp_min(1.0e-12).log()
).sum(dim=-1)
entropy_norm = entropy / torch.log(torch.tensor(17.0))
assert torch.allclose(features[..., 0], top[..., 0], atol=1.0e-6)
assert torch.allclose(
features[..., 1],
top[..., 0] - top[..., 1],
atol=1.0e-6,
)
assert torch.allclose(features[..., 2], entropy_norm, atol=1.0e-6)
assert torch.allclose(
surface(hidden, logits),
surface.forward_features(hidden, features),
)
projected_hidden = surface.hidden_proj(
hidden.to(dtype=surface.hidden_proj.weight.dtype)
).float()
assert torch.allclose(
surface(hidden, logits),
surface.forward_projected_features(projected_hidden, features),
)
def test_native_confidence_observes_the_model_owned_candidate_token() -> None:
cfg = NexumConfig.tiny()
head = CapacityLMHead(
torch.nn.Linear(cfg.hidden_size, cfg.vocab_size, bias=False),
cfg,
n_layers=1,
)
token_bridge = NexumTBR(cfg)
hidden = torch.randn(1, 1, cfg.hidden_size)
first_logits = torch.full((1, 1, cfg.vocab_size), -10.0)
second_logits = first_logits.clone()
first_logits[..., 3] = 10.0
second_logits[..., 7] = 10.0
first = head.native_decode_confidence_hidden_t(
hidden,
first_logits,
token_bridge,
)
second = head.native_decode_confidence_hidden_t(
hidden,
second_logits,
token_bridge,
)
assert first.shape == hidden.shape
assert second.shape == hidden.shape
assert torch.isfinite(first).all()
assert torch.isfinite(second).all()
assert not torch.equal(first, second)
torch.testing.assert_close(
first.float().pow(2).mean(dim=-1).sqrt(),
hidden.float().pow(2).mean(dim=-1).sqrt(),
atol=1.0e-5,
rtol=1.0e-5,
)
def test_native_confidence_training_accepts_frozen_routed_features() -> None:
cfg = NexumConfig.tiny()
surface = NativeDecodeConfidenceSurface(cfg)
with torch.no_grad():
surface.complete_head.weight.normal_(mean=0.0, std=0.005)
surface.fragment_head.weight.copy_(-surface.complete_head.weight)
hidden = torch.randn(1, 5, cfg.hidden_size)
logits = torch.randn(1, 5, cfg.vocab_size)
labels = torch.tensor([[-100, 11, 12, 13, 14]])
verified_completion_t = torch.tensor([[False, False, True, True]])
features = surface._logit_features(logits[:, :-1, :])
projected_hidden = surface.hidden_proj(
hidden[:, :-1, :].to(dtype=surface.hidden_proj.weight.dtype)
).float()
direct_loss, _direct_receipt = native_decode_confidence_training_loss(
surface,
hidden,
logits,
labels,
verified_completion_t=verified_completion_t,
)
cached_loss, cached_receipt = native_decode_confidence_training_loss(
surface,
hidden,
logits,
labels,
verified_completion_t=verified_completion_t,
logit_features=features,
projected_hidden=projected_hidden,
)
live_projection_loss, live_projection_receipt = (
native_decode_confidence_training_loss(
surface,
hidden,
logits,
labels,
verified_completion_t=verified_completion_t,
logit_features=features,
)
)
torch.testing.assert_close(cached_loss, direct_loss)
torch.testing.assert_close(live_projection_loss, direct_loss)
live_projection_loss.backward()
assert surface.hidden_proj.weight.grad is not None
assert torch.count_nonzero(surface.hidden_proj.weight.grad) > 0
assert cached_receipt["native_decode_confidence_positive"] == 1
assert cached_receipt["native_decode_confidence_negative"] == 2
assert live_projection_receipt["native_decode_confidence_positive"] == 1
assert live_projection_receipt["native_decode_confidence_negative"] == 2
def test_native_decode_analysis_exposes_confidence_and_rbo_count() -> None:
cfg = NexumConfig.tiny()
head = CapacityLMHead(
torch.nn.Linear(cfg.hidden_size, cfg.vocab_size, bias=False),
cfg,
n_layers=1,
)
confidence = torch.tensor([[[0.8, 0.2, 0.5, 1.4]]])
object.__setattr__(head, "_last_native_decode_confidence_t", confidence)
head._rbo_generation_forward_count_t.fill_(3)
analysis = head.native_decode_analysis_t()
assert analysis is not None
torch.testing.assert_close(
analysis,
torch.tensor([[0.8, 0.2, 0.5, 1.4, 3.0, 0.0, 0.0, 0.0, 0.0, 0.0]]),
)
def test_selected_token_recorder_is_idempotent_for_each_snapshot() -> None:
observed: list[torch.Tensor] = []
owner = SimpleNamespace(
record_generation_output_token_ids_t=lambda value: observed.append(
value.detach().clone()
)
)
recorder = _NativeSelectedTokenRecorder(owner, physical_tokens=2)
first = torch.tensor([[1, 2, 7]], dtype=torch.long)
second = torch.tensor([[1, 2, 7, 8]], dtype=torch.long)
recorder(first, scores=None)
recorder(first, scores=None)
recorder(second, scores=None)
assert recorder.recorded_tokens == 2
assert len(observed) == 2
assert torch.equal(observed[0], torch.tensor([[7]]))
assert torch.equal(observed[1], torch.tensor([[8]]))
def test_completion_context_uses_ordered_tokens_intent_and_action() -> None:
cfg = NexumConfig.tiny()
head = CapacityLMHead(
torch.nn.Linear(cfg.hidden_size, cfg.vocab_size, bias=False),
cfg,
n_layers=1,
)
token_bridge = NexumTBR(cfg)
object.__setattr__(
head,
"_request_completion_intent_hidden_t",
torch.ones(1, 1, cfg.hidden_size),
)
hidden = torch.zeros(1, 3, cfg.hidden_size)
token_ids = torch.tensor([[3, 4, 5]], dtype=torch.long)
action = torch.full_like(hidden, 0.5)
packet = head.completion_context_sequence_t(
hidden,
token_ids,
token_bridge,
action,
)
assert packet.hidden_t.shape == hidden.shape
assert torch.isfinite(packet.hidden_t).all()
assert torch.all(packet.delta_norm_t > 0)
torch.testing.assert_close(
packet.token_mass_t,
torch.tensor([[[1.0], [2.0], [3.0]]]),
)
def test_templates_are_generic_and_uncapped() -> None:
names = template_names()
assert {
"tool_task",
"self_correction",
"terminal",
"code_change",
"research",
"transaction",
"reproduction",
"triage",
"disclosure",
"patching",
} <= set(names)
rendered = render_template("tool_task", user_task="inspect and verify")
assert "Nexum" in rendered
assert "until the objective is complete" in rendered
def test_cli_doctor_json() -> None:
process = subprocess.run(
[
sys.executable,
"-m",
"nexum_runtime.cli",
"doctor",
"--model",
str(MODEL_DIR),
],
check=True,
text=True,
capture_output=True,
env=_source_cli_environment(),
)
payload = json.loads(process.stdout)
assert payload["ok"] is True
assert payload["name"] == "Nexum"
def test_cli_doctor_fails_without_model() -> None:
environment = _source_cli_environment()
environment.pop("NEXUM_MODEL_DIR", None)
process = subprocess.run(
[sys.executable, "-m", "nexum_runtime.cli", "doctor"],
check=False,
text=True,
capture_output=True,
env=environment,
)
assert process.returncode == 1
assert json.loads(process.stdout)["ok"] is False
def test_cli_exec_smoke_prepares_workspace(tmp_path: Path) -> None:
workspace = tmp_path / "new-workspace"
process = subprocess.run(
[
sys.executable,
"-m",
"nexum_runtime.cli",
"tools",
"exec-smoke",
"--workspace",
str(workspace),
],
check=True,
text=True,
capture_output=True,
env=_source_cli_environment(),
)
payload = json.loads(process.stdout)
assert payload["ok"] is True
assert workspace.is_dir()
assert "nexum-tool-ok" in payload["results"][0]["output"]
def test_real_bundle_map_is_complete() -> None:
report = validate_bundle(MODEL_DIR, deep=False)
assert report.ok is True
assert report.tensor_count == 113
assert report.load_map_ok is True
assert report.tensor_bytes > 20 * 1024**3
def test_generation_route_strength_is_owned_by_numbered_state() -> None:
state = load_file(
str(MODEL_DIR / "safetensors" / "000004.safetensors"),
device="cpu",
)
key = "000531"
assert sorted(state) == [f"{index:06d}" for index in range(1, 666)]
cfg = NexumConfig(hidden_size=8, hop_mlp_hidden=8)
surface = NativeDecodeConfidenceSurface(cfg)
surface.load_state_dict({"route_strength_logit": state[key]}, strict=False)
strength = surface.route_strength(torch.ones((), dtype=torch.float32))
assert torch.allclose(
strength,
torch.tensor(cfg.capacity_delta_rms_ratio),
atol=1.0e-6,
)
def test_numbered_state_merges_every_factor_into_one_model(tmp_path: Path) -> None:
model = torch.nn.Sequential(torch.nn.Linear(2, 2, bias=False))
linear = cast(torch.nn.Linear, model[0])
with torch.no_grad():
linear.weight.zero_()
input_factor = torch.tensor([[1.0, 2.0]])
output_factor = torch.tensor([[3.0], [4.0]])
state_path = tmp_path / "000002.safetensors"
save_file(
{"000001": input_factor, "000002": output_factor},
str(state_path),
)
bindings = [
{
"id": "000001",
"module": [0],
"shape": [2, 2],
"tensors": ["000001", "000002"],
}
]
graph_sha256 = hashlib.sha256(
json.dumps(
[{"module": [0], "shape": [2, 2]}],
sort_keys=True,
separators=(",", ":"),
).encode()
).hexdigest()
receipt = _merge_numbered_state(
model,
state_path,
{
"bindings": bindings,
"coefficient": [2, 1],
"graph_sha256": graph_sha256,
"schema": "nexum.state.v2",
},
)
assert receipt == {
"complete": True,
"state_tensors": 2,
"state_tensors_bound": 2,
"weight_bindings": 1,
}
assert torch.equal(
linear.weight,
torch.tensor([[6.0, 12.0], [8.0, 16.0]]),
)
def test_real_numbered_state_has_only_numeric_keys_and_complete_bindings() -> None:
state = load_file(
str(MODEL_DIR / "safetensors" / "000002.safetensors"),
device="cpu",
)
config = json.loads((MODEL_DIR / "state_config.json").read_text(encoding="utf-8"))
bound = [
tensor_id for binding in config["bindings"] for tensor_id in binding["tensors"]
]
assert sorted(state) == [f"{index:06d}" for index in range(1, 177)]
assert sorted(bound) == sorted(state)
assert len(config["bindings"]) == 88
def test_real_numbered_state_targets_match_constructed_graph() -> None:
config = _architecture_config(MODEL_DIR / "config.json")
state_config = json.loads(
(MODEL_DIR / "state_config.json").read_text(encoding="utf-8")
)
with torch.device("meta"):
model = NexumForCausalLM(config)
assert _validate_numbered_state_targets(model, state_config) == 88
def test_context_intent_numbered_state_binds_complete_graph_partition() -> None:
config = _architecture_config(MODEL_DIR / "config.json")
with torch.device("meta"):
model = NexumForCausalLM(config)
names, shapes = _context_intent_package_contract(model)
assert len(names) == len(shapes) == 18
receipt = _load_numbered_context_intent(
model,
MODEL_DIR / "safetensors" / "000112.safetensors",
device="cpu",
dtype=torch.bfloat16,
)
assert receipt == {
"complete": True,
"state_tensors": 18,
"graph_tensors": 18,
}
parameters = dict(model.named_parameters())
assert all(parameters[name].device.type == "cpu" for name in names)
assert all(torch.isfinite(parameters[name]).all() for name in names)
gate_names = tuple(name for name in names if name.endswith("intent_pivot_gate"))
assert len(gate_names) == 6
assert all(
torch.equal(
parameters[name],
torch.full_like(parameters[name], -6.0),
)
for name in gate_names
)
def test_action_numbered_state_binds_complete_graph_partition() -> None:
config = _architecture_config(MODEL_DIR / "config.json")
with torch.device("meta"):
model = NexumForCausalLM(config)
names, shapes = _action_package_contract(model)
assert len(names) == len(shapes) == 18
receipt = _load_numbered_action(
model,
MODEL_DIR / "safetensors" / "000113.safetensors",
device="cpu",
dtype=torch.bfloat16,
)
assert receipt == {
"complete": True,
"state_tensors": 18,
"graph_tensors": 18,
}
parameters = dict(model.named_parameters())
assert all(parameters[name].device.type == "cpu" for name in names)
assert all(torch.isfinite(parameters[name]).all() for name in names)
gate_names = tuple(name for name in names if name.endswith("action_pivot_gate"))
assert len(gate_names) == 6
assert all(
torch.equal(
parameters[name],
torch.full_like(parameters[name], -6.0),
)
for name in gate_names
)
def test_release_tokenizer_loads_directly_and_preserves_exact_tool_text() -> None:
config = _architecture_config(MODEL_DIR / "config.json")
tokenizer = _load_tokenizer(MODEL_DIR, config)
text = "printf NEXUM_OK\n**/*.toml"
assert type(tokenizer._tokenizer).__module__ == "fastokens._compat"
assert (
tokenizer.decode(
tokenizer.encode(text, add_special_tokens=False),
skip_special_tokens=False,
)
== text
)
assert tokenizer.model_max_length == 4_194_304
assert len(tokenizer) == 128000
assert set(tokenizer.get_vocab().values()) == set(range(128000))
visible_ids = tokenizer.encode("verified", add_special_tokens=False) + [124895]
assert tokenizer.decode(visible_ids, skip_special_tokens=True) == "verified"
assert tokenizer.decode(visible_ids, skip_special_tokens=False).endswith(
"<|endoftext|>"
)
for token_id in range(125017, 128000):
rendered = tokenizer.decode([token_id], skip_special_tokens=True)
assert rendered
assert tokenizer.encode(rendered, add_special_tokens=False) == [token_id]
def test_context_admission_aggregates_every_source_region() -> None:
cfg = NexumConfig.tiny()
token_bridge = NexumTBR(cfg)
source_ids_t = (
torch.arange(4096, dtype=torch.long).remainder(cfg.vocab_size).reshape(1, -1)
)
focus_ids_t = torch.tensor([[7, 11, 13, 17]], dtype=torch.long)
baseline = token_bridge.context_admission(
source_ids_t,
focus_ids_t,
exact_token_budget=cfg.token_dim * 2,
)
assert baseline.aggregate_token_t.shape == (
1,
cfg.token_dim * 4 + 1,
cfg.token_dim,
)
assert int(baseline.source_token_count_t[0]) == source_ids_t.shape[1]
assert int(baseline.selected_token_count_t[0]) <= cfg.token_dim * 2
for position in (0, source_ids_t.shape[1] // 2, source_ids_t.shape[1] - 1):
changed_ids_t = source_ids_t.clone()
changed_ids_t[0, position] = (changed_ids_t[0, position] + 37).remainder(
cfg.vocab_size
)
changed = token_bridge.context_admission(
changed_ids_t,
focus_ids_t,
exact_token_budget=cfg.token_dim * 2,
)
assert not torch.equal(changed.aggregate_token_t, baseline.aggregate_token_t)
def test_context_admission_retains_exact_query_relevant_span() -> None:
cfg = NexumConfig.tiny()
token_bridge = NexumTBR(cfg)
marker_ids_t = torch.tensor(
[7, 11, 13, 17, 19, 23, 29, 31],
dtype=torch.long,
)
marker_query_t = F.normalize(
token_bridge.token_ids_to_token(marker_ids_t).mean(dim=0),
dim=-1,
)
candidates_t = torch.arange(cfg.vocab_size, dtype=torch.long)
candidate_score_t = torch.einsum(
"vd,d->v",
token_bridge.token_ids_to_token(candidates_t),
marker_query_t,
)
background_id = int(torch.argmin(candidate_score_t))
source_ids_t = torch.full((1, cfg.token_dim * 8), background_id)
marker_start = cfg.token_dim * 4 + 37
source_ids_t[0, marker_start : marker_start + marker_ids_t.shape[0]] = marker_ids_t
admitted = token_bridge.context_admission(
source_ids_t,
marker_ids_t.reshape(1, -1),
exact_token_budget=cfg.token_dim,
)
selected = admitted.selected_token_ids_t[0].tolist()
marker = marker_ids_t.tolist()
assert any(
selected[index : index + len(marker)] == marker
for index in range(len(selected) - len(marker) + 1)
)
def test_context_admission_pivots_to_tail_intent_in_repeated_long_prompt() -> None:
cfg = NexumConfig.tiny()
token_bridge = NexumTBR(cfg)
filler_id_t = torch.tensor([5], dtype=torch.long)
tail_intent_t = torch.tensor(
[7, 11, 13, 17, 19, 23, 29, 31],
dtype=torch.long,
)
source_ids_t = filler_id_t.repeat(cfg.token_dim * 32).reshape(1, -1)
source_ids_t[0, -tail_intent_t.shape[0] :] = tail_intent_t
admitted = token_bridge.context_admission(
source_ids_t,
source_ids_t,
exact_token_budget=cfg.token_dim,
)
selected = admitted.selected_token_ids_t[0].tolist()
marker = tail_intent_t.tolist()
assert any(
selected[index : index + len(marker)] == marker
for index in range(len(selected) - len(marker) + 1)
)
def test_eager_attention_online_softmax_matches_dense_short_sequence() -> None:
from nexum_runtime.architecture import (
DUAL_CHUNK_PRETRAIN_LENGTH,
NATIVE_ATTENTION_TILE_KEYS,
NexumAttention,
dual_chunk_position_ids,
eager_attention_forward,
)
from nexum_runtime.architecture_config import NexumArchitectureConfig
cfg = NexumArchitectureConfig(
hidden_size=32,
num_attention_heads=4,
num_key_value_heads=2,
max_position_embeddings=4_194_304,
)
module = NexumAttention(cfg, layer_idx=0)
query = torch.randn(1, 4, 3, 8)
key = torch.randn(1, 2, 11, 8)
value = torch.randn(1, 2, 11, 8)
dense_out, _ = eager_attention_forward(
module, query, key, value, None, module.scaling, dropout=0.0
)
# Force tiled composition on a short key axis.
import nexum_runtime.architecture as arch_mod
original = arch_mod.NATIVE_ATTENTION_TILE_KEYS
arch_mod.NATIVE_ATTENTION_TILE_KEYS = 4
try:
tiled_out, _ = eager_attention_forward(
module, query, key, value, None, module.scaling, dropout=0.0
)
finally:
arch_mod.NATIVE_ATTENTION_TILE_KEYS = original
torch.testing.assert_close(tiled_out, dense_out, atol=1e-5, rtol=1e-5)
assert cfg.max_position_embeddings == 4_194_304
assert NATIVE_ATTENTION_TILE_KEYS >= 1
absolute = torch.arange(
DUAL_CHUNK_PRETRAIN_LENGTH + 128, dtype=torch.long
).unsqueeze(0)
folded = dual_chunk_position_ids(absolute)
assert int(folded.amax()) < DUAL_CHUNK_PRETRAIN_LENGTH
torch.testing.assert_close(
dual_chunk_position_ids(torch.arange(9, dtype=torch.long).unsqueeze(0))[0],
torch.arange(9, dtype=torch.long),
)
beyond_advertised_context = torch.tensor(
[[4_194_304, 8_388_609, 16_777_219]],
dtype=torch.long,
)
beyond_folded = dual_chunk_position_ids(beyond_advertised_context)
assert beyond_folded.shape == beyond_advertised_context.shape
assert torch.all(beyond_folded.ge(0))
torch.testing.assert_close(
beyond_folded,
torch.remainder(
beyond_advertised_context,
DUAL_CHUNK_PRETRAIN_LENGTH,
),
)
distinct_late_chunk_offsets = torch.tensor(
[
[
DUAL_CHUNK_PRETRAIN_LENGTH + 17,
DUAL_CHUNK_PRETRAIN_LENGTH + 8_192 + 17,
]
],
dtype=torch.long,
)
distinct_late_chunk_phases = dual_chunk_position_ids(distinct_late_chunk_offsets)
assert distinct_late_chunk_phases[0, 0] != distinct_late_chunk_phases[0, 1]
assert not hasattr(arch_mod, "NATIVE_ATTENTION_POSITION_APERTURE")
def test_tiled_attention_dropout_preserves_online_softmax_normalizer(
monkeypatch: pytest.MonkeyPatch,
) -> None:
import nexum_runtime.architecture as arch_mod
from nexum_runtime.architecture import NexumAttention, eager_attention_forward
from nexum_runtime.architecture_config import NexumArchitectureConfig
cfg = NexumArchitectureConfig(
hidden_size=32,
num_attention_heads=4,
num_key_value_heads=2,
)
module = NexumAttention(cfg, layer_idx=0)
module.train()
query = torch.randn(1, 4, 3, 8)
key = torch.randn(1, 2, 11, 8)
value = torch.randn(1, 2, 11, 8)
dense_out, _ = eager_attention_forward(
module,
query,
key,
value,
None,
module.scaling,
dropout=0.0,
)
original_tile = arch_mod.NATIVE_ATTENTION_TILE_KEYS
monkeypatch.setattr(arch_mod, "NATIVE_ATTENTION_TILE_KEYS", 4)
monkeypatch.setattr(
F,
"dropout",
lambda weights, **_kwargs: weights * 2,
)
try:
dropped_out, _ = eager_attention_forward(
module,
query,
key,
value,
None,
module.scaling,
dropout=0.5,
)
finally:
arch_mod.NATIVE_ATTENTION_TILE_KEYS = original_tile
torch.testing.assert_close(
dropped_out,
dense_out * 2,
atol=1e-5,
rtol=1e-5,
)
def test_nexum_long_context_position_stack_compose_absolute_causal_and_dual_chunk_rope() -> (
None
):
from nexum_runtime.architecture import (
DUAL_CHUNK_LOCAL_SIZE,
DUAL_CHUNK_PRETRAIN_LENGTH,
NATIVE_POSITION_APERTURE,
build_long_context_position_stack,
dual_chunk_position_ids,
)
absolute = torch.arange(17, dtype=torch.long).unsqueeze(0).expand(2, -1)
stack = build_long_context_position_stack(absolute, chunk_size=64, local_size=8)
torch.testing.assert_close(stack.absolute_positions, absolute)
torch.testing.assert_close(stack.rope_position_ids[0], torch.arange(17))
assert stack.dual_chunk_intra.shape == absolute.shape
assert stack.dual_chunk_successive.shape == absolute.shape
assert stack.at_successive_seam.shape == absolute.shape
assert not stack.at_successive_seam.any()
seam_abs = torch.tensor([[67, 74]], dtype=torch.long)
seam_stack = build_long_context_position_stack(
seam_abs, chunk_size=64, local_size=8
)
assert bool(seam_stack.at_successive_seam[0, 0])
assert not bool(seam_stack.at_successive_seam[0, 1])
assert int(seam_stack.rope_position_ids[0, 0]) == 3
assert int(seam_stack.rope_position_ids[0, 1]) == 10
long_abs = torch.arange(
DUAL_CHUNK_PRETRAIN_LENGTH + DUAL_CHUNK_LOCAL_SIZE + 4,
dtype=torch.long,
).unsqueeze(0)
long_stack = build_long_context_position_stack(long_abs)
assert torch.equal(long_stack.absolute_positions, long_abs)
assert int(long_stack.rope_position_ids.amax()) < DUAL_CHUNK_PRETRAIN_LENGTH
assert int(long_stack.rope_position_ids[0, DUAL_CHUNK_PRETRAIN_LENGTH]) == 0
assert (
int(
long_stack.rope_position_ids[
0, DUAL_CHUNK_PRETRAIN_LENGTH + DUAL_CHUNK_LOCAL_SIZE
]
)
== DUAL_CHUNK_LOCAL_SIZE
)
torch.testing.assert_close(
dual_chunk_position_ids(long_abs),
long_stack.rope_position_ids,
)
short = dual_chunk_position_ids(12, device="cpu", batch=1, chunk_size=64)
torch.testing.assert_close(short[0], torch.arange(12, dtype=torch.long))
long_len = DUAL_CHUNK_PRETRAIN_LENGTH + DUAL_CHUNK_LOCAL_SIZE + 2048
assert long_len < NATIVE_POSITION_APERTURE
folded = dual_chunk_position_ids(long_len, device="cpu", batch=1)
assert int(folded.amax()) < DUAL_CHUNK_PRETRAIN_LENGTH
def test_configure_nexum_long_context_geometry_caps_yarn_factor() -> None:
from nexum_runtime.architecture import (
DUAL_CHUNK_PRETRAIN_LENGTH,
NATIVE_POSITION_APERTURE,
NEXUM_TRAINED_YARN_FACTOR,
configure_nexum_long_context_geometry,
)
from nexum_runtime.architecture_config import NexumArchitectureConfig
cfg = NexumArchitectureConfig(
max_position_embeddings=262_144,
rope_parameters={
"rope_theta": 1_000_000.0,
"rope_type": "yarn",
"factor": 16.0,
},
)
configure_nexum_long_context_geometry(cfg)
assert cfg.max_position_embeddings == NATIVE_POSITION_APERTURE
assert cfg.rope_parameters["rope_type"] == "yarn"
assert float(cfg.rope_parameters["factor"]) == NEXUM_TRAINED_YARN_FACTOR
assert (
int(cfg.rope_parameters["original_max_position_embeddings"])
== DUAL_CHUNK_PRETRAIN_LENGTH
)
default_cfg = NexumArchitectureConfig(
max_position_embeddings=131_072,
rope_parameters={"rope_theta": 1_000_000.0, "rope_type": "default"},
)
configure_nexum_long_context_geometry(default_cfg)
assert default_cfg.max_position_embeddings == NATIVE_POSITION_APERTURE
assert default_cfg.rope_parameters["rope_type"] == "default"
def test_compute_nexum_rope_parameters_yarn_differs_from_default() -> None:
from nexum_runtime.architecture import (
DUAL_CHUNK_PRETRAIN_LENGTH,
NEXUM_TRAINED_YARN_FACTOR,
NexumRotaryEmbedding,
compute_nexum_rope_parameters,
resolve_nexum_yarn_geometry,
)
from nexum_runtime.architecture_config import NexumArchitectureConfig
default_cfg = NexumArchitectureConfig(
hidden_size=64,
num_attention_heads=4,
num_key_value_heads=2,
rope_parameters={"rope_theta": 1_000_000.0, "rope_type": "default"},
)
yarn_cfg = NexumArchitectureConfig(
hidden_size=64,
num_attention_heads=4,
num_key_value_heads=2,
rope_parameters={
"rope_theta": 1_000_000.0,
"rope_type": "yarn",
"factor": 8.0,
"original_max_position_embeddings": DUAL_CHUNK_PRETRAIN_LENGTH,
},
)
resolved = resolve_nexum_yarn_geometry(yarn_cfg)
assert resolved["rope_type"] == "yarn"
assert float(resolved["factor"]) == NEXUM_TRAINED_YARN_FACTOR
default_inv, default_scale = compute_nexum_rope_parameters(default_cfg)
yarn_inv, yarn_scale = compute_nexum_rope_parameters(yarn_cfg)
assert default_scale == 1.0
assert yarn_scale > 1.0
assert not torch.allclose(default_inv, yarn_inv)
rotary = NexumRotaryEmbedding(yarn_cfg)
assert rotary.rope_type == "yarn"
assert rotary.attention_scaling == yarn_scale
torch.testing.assert_close(rotary.inv_freq, yarn_inv)
def test_materialize_runtime_buffers_preserves_yarn_rope_geometry() -> None:
from nexum_runtime.architecture import (
DUAL_CHUNK_PRETRAIN_LENGTH,
NEXUM_TRAINED_YARN_FACTOR,
NexumForCausalLM,
compute_nexum_rope_parameters,
)
from nexum_runtime.architecture_config import NexumArchitectureConfig
from nexum_runtime.runner import _materialize_runtime_buffers
cfg = NexumArchitectureConfig(
hidden_size=64,
num_attention_heads=4,
num_key_value_heads=2,
num_hidden_layers=2,
layer_types=["conv", "full_attention"],
rope_parameters={
"rope_theta": 1_000_000.0,
"rope_type": "yarn",
"factor": 12.0,
"original_max_position_embeddings": DUAL_CHUNK_PRETRAIN_LENGTH,
},
)
assert float(cfg.rope_parameters["factor"]) == 12.0
expected_inv, expected_scale = compute_nexum_rope_parameters(cfg)
with torch.device("meta"):
model = NexumForCausalLM(cfg)
_materialize_runtime_buffers(model, "cpu")
pos_emb = model.model.pos_emb
assert pos_emb.rope_type == "yarn"
assert float(cfg.rope_parameters["factor"]) == NEXUM_TRAINED_YARN_FACTOR
assert pos_emb.attention_scaling == expected_scale
torch.testing.assert_close(pos_emb.inv_freq, expected_inv)
def test_resolve_nexum_attention_implementation_forces_eager_for_long_keys() -> None:
from nexum_runtime.architecture import (
NATIVE_ATTENTION_TILE_KEYS,
resolve_nexum_attention_implementation,
)
short = resolve_nexum_attention_implementation("sdpa", key_len=128)
assert short == "sdpa"
long = resolve_nexum_attention_implementation(
"flash_attention_2",
key_len=NATIVE_ATTENTION_TILE_KEYS + 1,
)
assert long == "eager"
default_long = resolve_nexum_attention_implementation(
None,
key_len=NATIVE_ATTENTION_TILE_KEYS + 512,
)
assert default_long == "eager"
def test_eager_attention_tiled_causal_mask_matches_dense() -> None:
import nexum_runtime.architecture as arch_mod
from nexum_runtime.architecture import NexumAttention, eager_attention_forward
from nexum_runtime.architecture_config import NexumArchitectureConfig
cfg = NexumArchitectureConfig(
hidden_size=32,
num_attention_heads=4,
num_key_value_heads=2,
)
module = NexumAttention(cfg, layer_idx=0)
query = torch.randn(1, 4, 3, 8)
key = torch.randn(1, 2, 13, 8)
value = torch.randn(1, 2, 13, 8)
causal = torch.triu(
torch.full((1, 1, 3, 13), float("-inf")),
diagonal=1,
)
dense_out, _ = eager_attention_forward(
module,
query,
key,
value,
causal,
module.scaling,
dropout=0.0,
)
original_tile = arch_mod.NATIVE_ATTENTION_TILE_KEYS
arch_mod.NATIVE_ATTENTION_TILE_KEYS = 4
try:
tiled_out, _ = eager_attention_forward(
module,
query,
key,
value,
causal,
module.scaling,
dropout=0.0,
)
finally:
arch_mod.NATIVE_ATTENTION_TILE_KEYS = original_tile
torch.testing.assert_close(tiled_out, dense_out, atol=1e-5, rtol=1e-5)
def test_zero_argument_config_defaults_to_eager_attention() -> None:
from nexum_runtime.architecture_config import NexumArchitectureConfig
cfg = NexumArchitectureConfig()
assert cfg._attn_implementation == "eager"
assert cfg.max_position_embeddings == 4_194_304
def test_context_admission_streams_multi_million_token_signatures() -> None:
cfg = NexumConfig.tiny()
token_bridge = NexumTBR(cfg)
source_ids_t = (
torch.arange(2_000_003, dtype=torch.long)
.remainder(cfg.vocab_size)
.reshape(1, -1)
)
focus_ids_t = torch.tensor([[7, 11, 13, 17]], dtype=torch.long)
baseline = token_bridge.context_admission(
source_ids_t,
focus_ids_t,
exact_token_budget=0,
)
changed_ids_t = source_ids_t.clone()
changed_ids_t[0, 1_500_001] = (changed_ids_t[0, 1_500_001] + 19).remainder(
cfg.vocab_size
)
changed = token_bridge.context_admission(
changed_ids_t,
focus_ids_t,
exact_token_budget=0,
)
assert baseline.aggregate_token_t.shape == (
1,
cfg.token_dim * 4 + 1,
cfg.token_dim,
)
assert int(baseline.source_token_count_t[0]) == source_ids_t.shape[1]
assert torch.isfinite(baseline.aggregate_token_t).all()
assert not torch.equal(baseline.aggregate_token_t, changed.aggregate_token_t)
def test_oversized_prompt_uses_trained_context_admission_without_cropping() -> None:
cfg = NexumConfig.tiny()
token_bridge = NexumTBR(cfg)
class Tokenizer:
def apply_chat_template(
self, *_args: object, **_kwargs: object
) -> torch.Tensor:
return torch.tensor([[1, 7, 11, 13, 2]], dtype=torch.long)
class Model(torch.nn.Module):
def __init__(self) -> None:
super().__init__()
self.embedding = torch.nn.Embedding(cfg.vocab_size, cfg.hidden_size)
self.lm_head = SimpleNamespace(experts=tuple(range(4)))
self.lm_head.__dict__["_nexum_token_bridge"] = token_bridge
def get_input_embeddings(self) -> torch.nn.Embedding:
return self.embedding
input_ids_t = (
torch.arange(1400, dtype=torch.long).remainder(cfg.vocab_size).reshape(1, -1)
)
prepared = _prepare_generation_context(
Model(),
Tokenizer(),
[{"role": "user", "content": "Find the relevant evidence."}],
input_ids_t,
context_limit=1024,
)
assert prepared.long_context_active is True
assert prepared.source_tokens == 1400
assert prepared.retained_learning_tokens == 0
assert prepared.physical_tokens < 1024
assert prepared.aggregate_tokens == cfg.token_dim * 4 + 1
assert prepared.selected_tokens > 0
assert prepared.focus_tokens == 5
assert prepared.inputs_embeds_t is not None
assert prepared.inputs_embeds_t.shape == (
1,
prepared.physical_tokens,
cfg.hidden_size,
)
assert prepared.input_ids_t.shape == (1, prepared.physical_tokens)
assert prepared.request_context_token_t is not None
assert prepared.request_context_token_t.shape[1] == (
prepared.aggregate_tokens + prepared.selected_tokens + prepared.focus_tokens
)
assert prepared.request_context_token_t.shape[1] == prepared.physical_tokens
torch.testing.assert_close(
prepared.input_ids_t[:, : prepared.aggregate_tokens],
token_bridge.token_to_token_ids(
prepared.request_context_token_t[:, : prepared.aggregate_tokens]
),
)
torch.testing.assert_close(
prepared.input_ids_t[:, prepared.aggregate_tokens :],
token_bridge.token_to_token_ids(
prepared.request_context_token_t[:, prepared.aggregate_tokens :]
),
)
def test_oversized_focus_preserves_native_tail_instruction_embedding() -> None:
cfg = NexumConfig.tiny()
token_bridge = NexumTBR(cfg)
tail_token_id = 37
class Tokenizer:
def apply_chat_template(
self, *_args: object, **_kwargs: object
) -> torch.Tensor:
focus = torch.full((1, cfg.token_dim * 9), 5, dtype=torch.long)
focus[0, -1] = tail_token_id
return focus
class Model(torch.nn.Module):
def __init__(self) -> None:
super().__init__()
self.embedding = torch.nn.Embedding(cfg.vocab_size, cfg.hidden_size)
self.lm_head = SimpleNamespace(experts=tuple(range(4)))
self.lm_head.__dict__["_nexum_token_bridge"] = token_bridge
def get_input_embeddings(self) -> torch.nn.Embedding:
return self.embedding
input_ids_t = torch.full((1, cfg.token_dim * 9), 5, dtype=torch.long)
input_ids_t[0, -1] = tail_token_id
model = Model()
prepared = _prepare_generation_context(
model,
Tokenizer(),
[{"role": "user", "content": "large current task"}],
input_ids_t,
context_limit=cfg.token_dim * 6,
)
assert prepared.long_context_active is True
assert prepared.focus_tokens > 0
assert prepared.inputs_embeds_t is not None
assert prepared.input_ids_t.shape == (1, prepared.physical_tokens)
assert int(prepared.input_ids_t[0, -1]) == tail_token_id
torch.testing.assert_close(
prepared.inputs_embeds_t[0, -1],
model.embedding(torch.tensor(tail_token_id)),
)
def test_retained_learning_evidence_stays_off_physical_prompt_surface() -> None:
cfg = NexumConfig.tiny()
token_bridge = NexumTBR(cfg)
class Tokenizer:
def apply_chat_template(
self, *_args: object, **_kwargs: object
) -> torch.Tensor:
return torch.tensor([[1, 7, 2]], dtype=torch.long)
class Model(torch.nn.Module):
def __init__(self) -> None:
super().__init__()
self.embedding = torch.nn.Embedding(cfg.vocab_size, cfg.hidden_size)
self.lm_head = SimpleNamespace(experts=tuple(range(4)))
self.lm_head.__dict__["_nexum_token_bridge"] = token_bridge
def get_input_embeddings(self) -> torch.nn.Embedding:
return self.embedding
model = Model()
input_ids_t = torch.tensor([[1, 7, 11, 2]], dtype=torch.long)
prepared = _prepare_generation_context(
model,
Tokenizer(),
[{"role": "user", "content": "Use retained execution evidence."}],
input_ids_t,
context_limit=1024,
)
assert prepared.long_context_active is False
assert prepared.source_tokens == input_ids_t.shape[1]
assert prepared.retained_learning_tokens == 0
assert prepared.physical_tokens == input_ids_t.shape[1]
assert prepared.inputs_embeds_t is None
assert prepared.request_context_token_t is None
torch.testing.assert_close(prepared.input_ids_t, input_ids_t)
def test_prompt_accepts_nullable_assistant_tool_call_for_correction() -> None:
config = _architecture_config(MODEL_DIR / "config.json")
tokenizer = _load_tokenizer(MODEL_DIR, config)
messages: list[dict[str, Any]] = [
{"role": "user", "content": "Read settings.json."},
{
"role": "assistant",
"content": None,
"tool_calls": [
_tool_call("call-1", "Bash", {"command": "cat setting.json"})
],
},
{
"role": "tool",
"tool_call_id": "call-1",
"content": '{"executed":true,"return_code":1}',
},
]
tools: list[dict[str, Any]] = [
{
"type": "function",
"function": {
"name": "Bash",
"description": "Execute a shell command.",
"parameters": {
"type": "object",
"properties": {"command": {"type": "string"}},
"required": ["command"],
},
},
}
]
encoded = _prompt_from_messages(tokenizer, messages, tools)
assert encoded.ndim == 2
assert encoded.shape[-1] > 0
rendered = tokenizer.decode(encoded[0], skip_special_tokens=False)
assert (
"Use a supplied tool before answering whenever the objective depends on external "
"state, persistent state, a side effect, or verification."
) in rendered
assert "answer directly when no tool evidence is needed" in rendered
assert (
"Preserve user-supplied identifiers, numbers, paths, and quoted values exactly"
in rendered
)
assert (
"Before a consequential or terminal action, draft the intended action privately "
"and verify every explicit user constraint against current tool evidence."
in rendered
)
assert (
"A matching attribute alone does not establish the requested object, category, "
"identity, or outcome" in rendered
)
assert (
"When a tool action fails or has no effect, use the returned evidence to change "
"the next action, tool, or arguments" in rendered
)
assert (
"copy the selected schema's complete function name and every argument key exactly"
in rendered
)
def test_release_architecture_is_local_and_matches_primary_graph() -> None:
config = _architecture_config(MODEL_DIR / "config.json")
with torch.device("meta"):
model = NexumForCausalLM(config)
state = model.state_dict()
assert type(config).__module__ == "nexum_runtime.architecture_config"
assert type(model).__module__ == "nexum_runtime.architecture"
assert len(state) == 271
assert (
sum(
key.endswith(".feed_forward.experts.gate_up_proj")
or key.endswith(".feed_forward.experts.down_proj")
for key in state
)
== 44
)
assert model.model.pos_emb.inv_freq.device.type == "meta"
assert model.model.pos_emb.original_inv_freq.device.type == "meta"
_materialize_runtime_buffers(model, "cpu")
assert model.model.pos_emb.inv_freq.device.type == "cpu"
assert model.model.pos_emb.original_inv_freq.device.type == "cpu"
def test_zero_argument_config_supports_generation_introspection() -> None:
config = NexumArchitectureConfig()
assert len(config.layer_types) == config.num_hidden_layers == 24
assert config.to_diff_dict()["model_type"] == "nexum_native"
@pytest.mark.parametrize("path", ("/etc/hostname", "../outside.safetensors"))
def test_numbered_tensor_paths_cannot_leave_model_directory(
tmp_path: Path, path: str
) -> None:
tensor_map = {
"tensors": [{"id": "000001", "file": path}],
}
with pytest.raises(ValueError, match="model directory"):
_required_tensor_path(tmp_path, tensor_map, "000001")
def test_numbered_authority_contract_matches_complete_runtime_graph(
tmp_path: Path,
) -> None:
tensor_map = json.loads((MODEL_DIR / "tensor_map.json").read_text(encoding="utf-8"))
state_path = MODEL_DIR / "safetensors" / "000004.safetensors"
geometry = _authority_geometry_from_package(state_path)
base_config = _architecture_config(MODEL_DIR / "config.json")
manifest, page_root, session_id, summary = _numbered_none_page_view(
MODEL_DIR,
tensor_map,
tmp_path,
)
config = NexumCoreConfig(
hidden_size=base_config.hidden_size,
vocab_size=base_config.vocab_size,
num_layers=base_config.num_hidden_layers,
num_appended_layers=summary["parent_layers"],
use_live_weights=True,
use_shared_token_table=False,
)
config.axis_a_count = summary["axis_a_count"]
config.axis_b_count = summary["axis_b_count"]
config.experts_per_axis = summary["experts_per_axis"]
for name, value in geometry.__dict__.items():
setattr(config, name, value)
config.token_tbr_checkpoint = str(MODEL_DIR / "safetensors" / "000003.safetensors")
token_bridge, proof = load_token_bridge(config, device="cpu", strict=True)
assert proof["checkpoint_loaded"] is True
with torch.device("meta"):
model = NexumForCausalLM(base_config)
wire_full_authority(
model,
config,
layer_count=summary["parent_layers"],
token_bridge=token_bridge,
authority_state=None,
)
names, expected_shapes = _authority_package_contract(model)
assert len(names) == 665
with open_tensor_package(state_path, semantic_names=names) as handle:
actual_shapes = tuple(
tuple(int(value) for value in handle.get_slice(name).get_shape())
for name in handle.keys()
)
assert actual_shapes == expected_shapes
model.lm_head.configure_none_paging_boundary(
manifest,
page_root,
session_id,
"cpu",
)
paged_names, paged_shapes = _authority_package_contract(model)
assert paged_names == names
assert paged_shapes == expected_shapes
def test_numbered_primary_loader_packs_native_expert_authority(
tmp_path: Path,
) -> None:
class PackedAuthority(torch.nn.Module):
def __init__(self) -> None:
super().__init__()
self.gate_up_proj = torch.nn.Parameter(torch.empty(2, 4, 2))
self.down_proj = torch.nn.Parameter(torch.empty(2, 2, 2))
class FeedForward(torch.nn.Module):
def __init__(self) -> None:
super().__init__()
self.experts = PackedAuthority()
class Layer(torch.nn.Module):
def __init__(self) -> None:
super().__init__()
self.feed_forward = FeedForward()
class Body(torch.nn.Module):
def __init__(self) -> None:
super().__init__()
self.embed_tokens = torch.nn.Embedding(3, 2)
self.layers = torch.nn.ModuleList([Layer()])
class Model(torch.nn.Module):
def __init__(self) -> None:
super().__init__()
self.config = SimpleNamespace(num_experts=2)
self.model = Body()
self.lm_head = torch.nn.Linear(2, 3, bias=False)
def get_input_embeddings(self) -> torch.nn.Module:
return self.model.embed_tokens
def get_output_embeddings(self) -> torch.nn.Module:
return self.lm_head
def tie_weights(self) -> None:
self.lm_head.weight = self.model.embed_tokens.weight
source = {
"model.embed_tokens.weight": torch.arange(6, dtype=torch.float32).reshape(3, 2),
}
for expert_index in range(2):
prefix = f"model.layers.0.feed_forward.experts.{expert_index}"
source[f"{prefix}.w1.weight"] = torch.full((2, 2), float(expert_index + 1))
source[f"{prefix}.w2.weight"] = torch.full((2, 2), float(expert_index + 3))
source[f"{prefix}.w3.weight"] = torch.full((2, 2), float(expert_index + 5))
primary_path = tmp_path / "000001.safetensors"
ordered_source = sorted(source.items())
save_file(
{
f"{index:06d}": value
for index, (_name, value) in enumerate(ordered_source, 1)
},
str(primary_path),
)
model = Model().to(device="meta")
receipt = _load_numbered_primary(
model,
primary_path,
device="cpu",
dtype=torch.float32,
)
assert receipt == {
"complete": True,
"source_tensors": 7,
"graph_tensors": 4,
"packed_tensors": 2,
}
loaded_layer = cast(Layer, model.model.layers[0])
assert torch.equal(
loaded_layer.feed_forward.experts.gate_up_proj[0],
torch.tensor([[1.0, 1.0], [1.0, 1.0], [5.0, 5.0], [5.0, 5.0]]),
)
assert model.lm_head.weight.data_ptr() == model.model.embed_tokens.weight.data_ptr()
def test_numbered_reader_maps_structural_package_without_metadata(
tmp_path: Path,
) -> None:
path = tmp_path / "000018.safetensors"
save_file(
{
"000001": torch.tensor(4),
"000002": torch.tensor([1, 2, 3, 4]),
},
str(path),
)
with open_tensor_package(path) as handle:
assert handle.keys() == ["accepted_generation_t", "session_id_t"]
assert torch.equal(handle.get_tensor("accepted_generation_t"), torch.tensor(4))
assert torch.equal(
handle.get_tensor("session_id_t"), torch.tensor([1, 2, 3, 4])
)
def test_frozen_catalog_reuses_route_device_dtype_materialization(
tmp_path: Path,
) -> None:
layer_path = tmp_path / "layer.safetensors"
fc1 = torch.arange(48, dtype=torch.float32).reshape(2, 8, 3)
fc2 = torch.arange(24, dtype=torch.float32).reshape(2, 3, 4)
save_file(
{
"expert_fc1_weight": fc1,
"expert_fc2_weight": fc2,
},
str(layer_path),
)
manifest = tmp_path / "manifest.json"
manifest.write_text(
json.dumps(
{
"none_layers": [{"layer": 0, "path": layer_path.name}],
}
),
encoding="utf-8",
)
catalog = NoNEFrozenExpertCatalogBoundary.from_manifest(manifest)
request = NoNEPageRequestPacket(
session_id_t=torch.tensor([1], dtype=torch.long),
generation_t=torch.tensor(1, dtype=torch.long),
route_pair_ids_t=torch.tensor([0, 1], dtype=torch.long),
unique_route_pair_ids_t=torch.tensor([0, 1], dtype=torch.long),
route_position_t=torch.tensor([0, 1], dtype=torch.long),
priority_t=torch.ones(2),
)
first = catalog.load(request, device="cpu", dtype=torch.bfloat16)
second = catalog.load(request, device="cpu", dtype=torch.bfloat16)
full_precision = catalog.load(request, device="cpu", dtype=torch.float32)
after_source_read = catalog.load(request, device="cpu", dtype=torch.bfloat16)
assert first.fc1_weight_t.dtype == torch.bfloat16
assert first.fc2_weight_t.dtype == torch.bfloat16
assert first.fc1_weight_t.data_ptr() == second.fc1_weight_t.data_ptr()
assert first.fc2_weight_t.data_ptr() == second.fc2_weight_t.data_ptr()
assert full_precision.fc1_weight_t.dtype == torch.float32
assert full_precision.fc2_weight_t.dtype == torch.float32
assert full_precision.fc1_weight_t.data_ptr() != first.fc1_weight_t.data_ptr()
assert after_source_read.fc1_weight_t.data_ptr() == first.fc1_weight_t.data_ptr()
assert after_source_read.fc2_weight_t.data_ptr() == first.fc2_weight_t.data_ptr()
def test_inference_active_pages_keep_compute_dtype_and_reuse_generation(
tmp_path: Path,
) -> None:
route_ids_t = torch.tensor([0, 1], dtype=torch.long)
state: dict[str, torch.Tensor] = {"step_t": torch.zeros(2, dtype=torch.long)}
for name in NoNEActivePageParameters._PARAMETER_NAMES:
state[name] = torch.randn(2, 2)
state[f"optimizer_mean_{name}"] = torch.zeros(2, 2)
state[f"optimizer_square_{name}"] = torch.zeros(2, 2)
active = NoNEActivePageParameters(
state,
route_ids_t,
device="cpu",
parameter_dtype=torch.bfloat16,
)
assert all(parameter.dtype == torch.bfloat16 for parameter in active.parameters())
assert active.optimizer_mean_fc1_down_t.dtype == torch.float32
store = NoNECapacityPageStoreBoundary(tmp_path)
session_id_t = torch.tensor([1, 2, 3, 4], dtype=torch.long)
store.begin_session(session_id_t)
store._active_materialization_key = (
0,
(0, 1),
torch.device("cpu"),
torch.bfloat16,
)
store._active_materialization = active
request = NoNEPageRequestPacket(
session_id_t=session_id_t,
generation_t=torch.tensor(1, dtype=torch.long),
route_pair_ids_t=route_ids_t,
unique_route_pair_ids_t=route_ids_t,
route_position_t=route_ids_t,
priority_t=torch.ones(2),
)
frozen = NoNEFrozenPageBatch(
fc1_weight_t=torch.empty(0),
fc2_weight_t=torch.empty(0),
route_pair_ids_t=route_ids_t,
catalog_revision_t=torch.zeros(32, dtype=torch.uint8),
)
reused = store.load_active_pages(
request,
frozen,
device="cpu",
parameter_dtype=torch.bfloat16,
reuse_materialization=True,
)
assert reused is active
def test_active_page_mutations_do_not_alias_durable_cache_state(
tmp_path: Path,
) -> None:
route_ids_t = torch.tensor([0], dtype=torch.long)
state: dict[str, torch.Tensor] = {"step_t": torch.zeros(1, dtype=torch.long)}
for name in NoNEActivePageParameters._PARAMETER_NAMES:
state[name] = torch.ones(1, 2, dtype=torch.float32)
state[f"optimizer_mean_{name}"] = torch.zeros(1, 2)
state[f"optimizer_square_{name}"] = torch.zeros(1, 2)
durable_parameter = state["fc1_down_t"].clone()
durable_step = state["step_t"].clone()
active = NoNEActivePageParameters(state, route_ids_t, device="cpu")
with torch.no_grad():
active.fc1_down_t.add_(5.0)
active.step_t.add_(1)
assert torch.equal(state["fc1_down_t"], durable_parameter)
assert torch.equal(state["step_t"], durable_step)
store = NoNECapacityPageStoreBoundary(tmp_path)
store.begin_session(torch.tensor([9, 8, 7, 6], dtype=torch.long))
store._page_file_cache[(0, "cached")] = {"value": torch.ones(1)}
store.restore_accepted_generation_boundary(torch.zeros((), dtype=torch.long))
assert store._page_file_cache == {}
def test_complete_catalog_route_avoids_dynamic_unique_and_hot_scalar_reads() -> None:
hidden_size = 4
state = {
"axis_a_weight": torch.randn(2, hidden_size),
"axis_b_weight": torch.randn(2, hidden_size),
"slot_weight": torch.randn(3, hidden_size),
"norm_weight": torch.ones(hidden_size),
"norm_bias": torch.zeros(hidden_size),
"router_top_k_t": torch.tensor([12], dtype=torch.long),
}
router = NexumNoNEFrozenRouterLayer(state, route_offset=24)
hidden = torch.randn(3, hidden_size)
expert_bias = torch.zeros(3, 12)
scalar_sync = AssertionError("router extracted a tensor scalar during forward")
with mock.patch.object(torch.Tensor, "__int__", side_effect=scalar_sync):
route = router(hidden, expert_bias)
with mock.patch(
"nexum_core.none_paged_layer.torch.unique",
side_effect=AssertionError("complete-catalog route used dynamic unique"),
):
unique_t, position_t = _route_request_index(
route.route_pair_ids_t,
router.catalog_route_pair_ids_t,
)
assert route.route_pair_ids_t.shape == (3, 12)
assert torch.equal(unique_t, torch.arange(24, 36, dtype=torch.long))
assert torch.equal(
unique_t.index_select(0, position_t),
route.route_pair_ids_t.reshape(-1),
)
def test_frozen_router_preserves_mass_for_every_selected_nonfinite_route() -> None:
hidden_size = 4
state = {
"axis_a_weight": torch.randn(2, hidden_size),
"axis_b_weight": torch.randn(2, hidden_size),
"slot_weight": torch.randn(3, hidden_size),
"norm_weight": torch.ones(hidden_size),
"norm_bias": torch.zeros(hidden_size),
"router_top_k_t": torch.tensor([12], dtype=torch.long),
}
router = NexumNoNEFrozenRouterLayer(state, route_offset=24)
expert_bias = torch.full((2, 12), float("nan"))
expert_bias[:, 0] = float("inf")
expert_bias[:, 1] = float("-inf")
route = router(torch.randn(2, hidden_size), expert_bias)
assert route.route_pair_ids_t.shape == (2, 12)
assert torch.isfinite(route.route_weight_t).all()
assert torch.isfinite(route.route_probability_t).all()
assert torch.all(route.route_weight_t > 0)
assert torch.all(route.route_probability_t > 0)
torch.testing.assert_close(
route.route_weight_t.sum(dim=-1),
torch.ones(2),
)
torch.testing.assert_close(
route.route_probability_t.sum(dim=-1),
torch.ones(2),
)
def test_preinitialized_route_pages_skip_parent_and_language_reload() -> None:
request = NoNEPageRequestPacket(
session_id_t=torch.tensor([1], dtype=torch.long),
generation_t=torch.tensor(1, dtype=torch.long),
route_pair_ids_t=torch.tensor([0, 1], dtype=torch.long),
unique_route_pair_ids_t=torch.tensor([0, 1], dtype=torch.long),
route_position_t=torch.tensor([0, 1], dtype=torch.long),
priority_t=torch.ones(2),
)
page_initialized = mock.Mock(return_value=True)
initialize_page = mock.Mock(
side_effect=AssertionError("preinitialized route was rewritten")
)
catalog_load = mock.Mock(
side_effect=AssertionError("preinitialized route reloaded its parent")
)
boundary = SimpleNamespace(
store=SimpleNamespace(
page_initialized_boundary=page_initialized,
initialize_page=initialize_page,
),
catalog=SimpleNamespace(load=catalog_load),
router=SimpleNamespace(expert_count=2),
manifest_path=Path("unused-manifest.json"),
bank_layer_idx=0,
)
with mock.patch(
"nexum_core.none_paged_layer.load_nexum_language_seed_boundary",
side_effect=AssertionError("preinitialized route reloaded its language state"),
):
NexumNoNEPagedLayerBoundary._initialize_selected_pages_boundary(
cast(Any, boundary),
request,
capacity_rank=2,
bit_language_dim=4,
)
assert page_initialized.call_count == 2
initialize_page.assert_not_called()
catalog_load.assert_not_called()
def test_fresh_session_inherits_release_base_and_overlays_touched_routes(
tmp_path: Path,
) -> None:
release_base = tmp_path / "release_base"
release_base.mkdir()
save_file(
{
"accepted_generation_t": torch.tensor([4]),
"session_id_t": torch.tensor([0, 0, 0, 0]),
},
str(release_base / "session_state.safetensors"),
)
base_generation = release_base / "generation_00000004.safetensors"
save_file(
{
"generation_t": torch.tensor([4]),
"route_pair_ids_t": torch.tensor([7, 8]),
},
str(base_generation),
)
store = NoNECapacityPageStoreBoundary(tmp_path)
session_id_t = torch.tensor([1, 2, 3, 4])
assert torch.equal(store.begin_session(session_id_t), torch.tensor(4))
assert store._page_source_by_route == {
7: (base_generation, 0),
8: (base_generation, 1),
}
assert store.page_initialized_boundary(7) is True
assert store.page_initialized_boundary(9) is False
session_root = (
tmp_path
/ hashlib.sha256(
session_id_t.contiguous().view(torch.uint8).numpy().tobytes()
).hexdigest()
)
session_generation = session_root / "generation_00000005.safetensors"
save_file(
{
"generation_t": torch.tensor([5]),
"route_pair_ids_t": torch.tensor([8]),
},
str(session_generation),
)
save_file(
{
"accepted_generation_t": torch.tensor([5]),
"session_id_t": session_id_t,
},
str(session_root / "session_state.safetensors"),
)
assert torch.equal(store.begin_session(session_id_t), torch.tensor(5))
assert store._page_source_by_route == {
7: (base_generation, 0),
8: (session_generation, 0),
}
def test_session_owned_none_banks_grow_without_ceiling_and_resume(
tmp_path: Path,
) -> None:
hidden_size = 4
token_size = 4
expert_count = 3
bank_count = 3
rows: list[dict[str, object]] = []
for bank_idx in range(bank_count):
generator = torch.Generator().manual_seed(100 + bank_idx)
layer_path = tmp_path / f"bank_{bank_idx}.safetensors"
save_file(
{
"expert_fc1_weight": torch.randn(
expert_count,
6,
hidden_size,
generator=generator,
),
"expert_fc2_weight": torch.randn(
expert_count,
hidden_size,
3,
generator=generator,
),
"axis_a_weight": torch.randn(1, hidden_size, generator=generator),
"axis_b_weight": torch.randn(
expert_count,
hidden_size,
generator=generator,
),
"slot_weight": torch.randn(1, hidden_size, generator=generator),
"norm_weight": torch.ones(hidden_size),
"norm_bias": torch.zeros(hidden_size),
"router_top_k_t": torch.tensor([expert_count], dtype=torch.long),
"page_expert_to_token_weight": torch.eye(token_size, hidden_size)
.unsqueeze(0)
.repeat(expert_count, 1, 1),
"page_token_to_expert_weight": torch.eye(hidden_size, token_size)
.unsqueeze(0)
.repeat(expert_count, 1, 1),
"page_route_embedding": torch.randn(
expert_count,
hidden_size,
generator=generator,
),
"bit_to_token_weight": torch.eye(token_size),
"token_to_bit_weight": torch.eye(token_size),
},
str(layer_path),
)
rows.append({"path": layer_path.name, "layer": bank_idx})
manifest = tmp_path / "manifest.json"
manifest.write_text(json.dumps({"none_layers": rows}), encoding="utf-8")
page_root = tmp_path / "pages"
session_id_t = torch.tensor([1, 2, 3, 4], dtype=torch.long)
hidden_t = torch.randn(2, hidden_size)
bit_t = torch.randn(2, token_size)
semantic_evidence_t = torch.tensor([[1.0, 0.0, 0.0, 0.0], [1.0, 0.25, 0.0, 0.0]])
stack = NexumNoNEPagedStack(manifest, page_root, "cpu")
stack.begin_session(session_id_t)
first = stack(
hidden_t,
bit_t,
torch.zeros(bank_count, 2, expert_count),
torch.ones(bank_count),
capacity_rank=2,
)
knowledge_before: dict[tuple[int, int], torch.Tensor] = {}
for bank_idx, layer_module in enumerate(stack.layers):
layer = cast(NexumNoNEPagedLayerBoundary, layer_module)
for local_expert_idx in range(expert_count):
route_id = bank_idx * expert_count + local_expert_idx
_session_t, _generation_t, page_state = (
layer.store._page_state_snapshot_boundary(route_id)
)
knowledge_before[(bank_idx, local_expert_idx)] = page_state[
"knowledge_transfer_t"
].clone()
failed_store = cast(NexumNoNEPagedLayerBoundary, stack.layers[1]).store
with (
mock.patch.object(
failed_store,
"commit",
side_effect=OSError("bank commit failed"),
),
pytest.raises(OSError, match="bank commit failed"),
):
stack.observe_grounded_outcome(
first.actual_route_pair_ids_t,
first.actual_route_weight_t,
first.draft_acceptance_t,
task_outcome_target_t=torch.zeros(()),
verification_target_t=torch.ones(()),
adaptation_t=torch.full((), 0.5),
grounded_evidence_token_t=semantic_evidence_t,
)
assert int(stack.accepted_generation_boundary()) == 0
assert all(
int(
cast(
NexumNoNEPagedLayerBoundary, layer
).store.accepted_generation_boundary()
)
== 0
for layer in stack.layers
)
outcome = stack.observe_grounded_outcome(
first.actual_route_pair_ids_t,
first.actual_route_weight_t,
first.draft_acceptance_t,
task_outcome_target_t=torch.zeros(()),
verification_target_t=torch.ones(()),
adaptation_t=torch.full((), 0.5),
grounded_evidence_token_t=semantic_evidence_t,
)
semantic_direction_t = F.normalize(semantic_evidence_t.mean(dim=0), dim=0)
alignment_before: list[torch.Tensor] = []
alignment_after: list[torch.Tensor] = []
for bank_idx, layer_module in enumerate(stack.layers):
layer = cast(NexumNoNEPagedLayerBoundary, layer_module)
for local_expert_idx in range(expert_count):
route_id = bank_idx * expert_count + local_expert_idx
_session_t, _generation_t, page_state = (
layer.store._page_state_snapshot_boundary(route_id)
)
assert not torch.equal(
page_state["knowledge_transfer_t"],
knowledge_before[(bank_idx, local_expert_idx)],
)
alignment_before.append(
F.cosine_similarity(
knowledge_before[(bank_idx, local_expert_idx)].reshape(1, -1),
semantic_direction_t.reshape(1, -1),
).reshape(())
)
alignment_after.append(
F.cosine_similarity(
page_state["knowledge_transfer_t"].reshape(1, -1),
semantic_direction_t.reshape(1, -1),
).reshape(())
)
growth = stack.grow_from_model_pressure(
donor_bank_idx_t=torch.tensor(1),
model_pressure_t=torch.tensor(4.2),
)
assert int(outcome.generation_t) == 1
assert bool(outcome.semantic_evidence_applied_t) is True
assert int(outcome.semantic_evidence_token_count_t) == 2
assert torch.stack(alignment_after).mean() > torch.stack(alignment_before).mean()
assert int(growth.banks_added_t) == 4
assert len(stack.layers) == 7
assert torch.equal(
stack.source_bank_indices("cpu"),
torch.tensor([0, 1, 2, 1, 1, 1, 1]),
)
with (
mock.patch.object(
stack,
"_write_stack_session_state_boundary",
side_effect=OSError("topology write failed"),
),
pytest.raises(OSError, match="topology write failed"),
):
stack.grow_from_model_pressure(
donor_bank_idx_t=torch.tensor(1),
model_pressure_t=torch.tensor(5.2),
)
assert len(stack.layers) == 7
assert torch.equal(
stack.source_bank_indices("cpu"),
torch.tensor([0, 1, 2, 1, 1, 1, 1]),
)
second = stack(
hidden_t,
bit_t,
torch.zeros(7, 2, expert_count),
torch.ones(7),
capacity_rank=2,
correction_pressure_t=torch.ones(()),
release_confidence_t=torch.ones(2, 1),
)
assert second.actual_route_pair_ids_t.shape == (7, 2, expert_count)
assert second.internal_agent_workspace_t.shape == (
7,
2,
expert_count,
hidden_size,
)
assert second.internal_agent_challenge_t.shape == (7, 2, expert_count)
torch.testing.assert_close(
second.internal_agent_acceptance_t.sum(dim=-1),
torch.ones(7, 2),
)
torch.testing.assert_close(
second.internal_agent_correction_pressure_t,
torch.ones(7, 2),
)
donor_layer = cast(NexumNoNEPagedLayerBoundary, stack.layers[1])
grown_layer = cast(NexumNoNEPagedLayerBoundary, stack.layers[3])
_donor_session, _donor_generation, donor_state = (
donor_layer.store._page_state_snapshot_boundary(expert_count)
)
_grown_session, _grown_generation, grown_state = (
grown_layer.store._page_state_snapshot_boundary(bank_count * expert_count)
)
torch.testing.assert_close(
grown_state["knowledge_transfer_t"],
donor_state["knowledge_transfer_t"],
)
torch.testing.assert_close(grown_state["step_t"], donor_state["step_t"])
restored = NexumNoNEPagedStack(manifest, page_root, "cpu")
restored.begin_session(session_id_t)
assert len(restored.layers) == 7
assert int(restored.accepted_generation_boundary()) == 1
resumed = restored(
hidden_t,
bit_t,
torch.zeros(7, 2, expert_count),
torch.ones(7),
capacity_rank=2,
)
assert resumed.actual_route_pair_ids_t.shape[0] == 7
more_growth = restored.grow_from_model_pressure(
donor_bank_idx_t=torch.tensor(3),
model_pressure_t=torch.tensor(9.1),
)
assert int(more_growth.banks_added_t) == 5
assert len(restored.layers) == 12
cold_reload = NexumNoNEPagedStack(manifest, page_root, "cpu")
cold_reload.begin_session(session_id_t)
assert len(cold_reload.layers) == 12
fresh_session = NexumNoNEPagedStack(manifest, page_root, "cpu")
fresh_session.begin_session(torch.tensor([5, 6, 7, 8], dtype=torch.long))
assert len(fresh_session.layers) == bank_count
source_t, donor_t, consumed_t = restored.shared_topology_boundary()
verified_future = NexumNoNEPagedStack(manifest, page_root, "cpu")
verified_future.configure_shared_topology_boundary(
source_t,
donor_t,
consumed_t,
)
verified_future.begin_session(torch.tensor([9, 10, 11, 12], dtype=torch.long))
assert len(verified_future.layers) == 12
assert int(verified_future.accepted_generation_boundary()) == 0
def test_session_state_isolated_restored_and_correlated(tmp_path: Path) -> None:
class PagedState(torch.nn.Module):
session_marker: torch.Tensor
def __init__(self) -> None:
super().__init__()
self.register_buffer("session_marker", torch.zeros(4, dtype=torch.long))
def begin_session(self, session_id_t: torch.Tensor) -> None:
self.session_marker.copy_(session_id_t)
class PathState(torch.nn.Module):
outcome: torch.Tensor
pro_score: torch.Tensor
anti_score: torch.Tensor
expert_identity_strength: torch.Tensor
layer_confidence: torch.Tensor
domain_cell_pro: torch.Tensor
domain_cell_anti: torch.Tensor
domain_growth_pressure: torch.Tensor
def __init__(self) -> None:
super().__init__()
self.register_buffer("outcome", torch.zeros(1))
self.register_buffer("pro_score", torch.zeros(2, 3))
self.register_buffer("anti_score", torch.zeros(2, 3))
self.register_buffer("expert_identity_strength", torch.zeros(2, 3))
self.register_buffer("layer_confidence", torch.zeros(2))
self.register_buffer("domain_cell_pro", torch.zeros(2, 2))
self.register_buffer("domain_cell_anti", torch.zeros(2, 2))
self.register_buffer("domain_growth_pressure", torch.zeros(2, 2))
def clear_self_correction_bias(self) -> None:
return
def clear_merkle_route(self) -> None:
return
class CorrectionState(Protocol):
trigger_success: torch.Tensor
trigger_failure: torch.Tensor
grounded_trigger_pressure: torch.Tensor
class Head(torch.nn.Module):
online_value: torch.Tensor
def __init__(self) -> None:
super().__init__()
self.register_buffer("online_value", torch.zeros(1))
correction = torch.nn.Module()
correction.register_buffer("trigger_success", torch.zeros(3))
correction.register_buffer("trigger_failure", torch.zeros(3))
correction.register_buffer(
"grounded_trigger_pressure", torch.zeros(3), persistent=False
)
self._self_correction_bank = cast(CorrectionState, correction)
self.traversal = torch.nn.Linear(1, 1, bias=False)
self.gate = torch.nn.Linear(1, 1, bias=False)
with torch.no_grad():
self.traversal.weight.zero_()
self.gate.weight.zero_()
self._path_registry = PathState()
self._none_paged_stack = PagedState()
self.experts = torch.nn.ModuleList()
class Model(torch.nn.Module):
def __init__(self) -> None:
super().__init__()
self.lm_head = Head()
state_root = tmp_path / "state"
model = Model()
bank = NexumSessionStateBank(model, device="cpu", state_root=state_root)
selected_call = {
"id": "call-a",
"type": "function",
"function": {
"name": "Read",
"arguments": json.dumps({"path": "selected.txt"}),
},
}
selected_observation = {
"name": "Read",
"args": {"path": "selected.txt"},
"ok": True,
"output": "selected",
"executed": True,
"tool_call_id": "call-a",
}
bank.activate("session-a")
marker_a = model.lm_head._none_paged_stack.session_marker.clone()
with torch.no_grad():
model.lm_head.online_value.fill_(7.0)
model.lm_head.traversal.weight.fill_(3.0)
model.lm_head._self_correction_bank.grounded_trigger_pressure[1] = 1.0
bank.set_pending_turn(
torch.tensor([0.75, 0.25]),
[selected_call],
[(1, torch.tensor([0, 2]), torch.tensor([0.25, 0.75]))],
torch.tensor([[[0, 1]], [[2, 3]]]),
torch.tensor([[[0.75, 0.25]], [[0.4, 0.6]]]),
torch.tensor([[True], [False]]),
torch.tensor(0.5),
prompt_fingerprint_t=torch.arange(32, dtype=torch.uint8),
)
bank.activate("session-b")
marker_b = model.lm_head._none_paged_stack.session_marker.clone()
assert not torch.equal(marker_a, marker_b)
assert torch.equal(model.lm_head.online_value, torch.zeros(1))
assert torch.equal(model.lm_head.traversal.weight, torch.zeros((1, 1)))
assert torch.equal(
model.lm_head._self_correction_bank.grounded_trigger_pressure,
torch.zeros(3),
)
with pytest.raises(ValueError, match="do not match"):
bank.pending_correction_state([selected_observation])
model.lm_head.online_value.fill_(11.0)
bank.activate("session-a")
assert torch.equal(model.lm_head._none_paged_stack.session_marker, marker_a)
assert torch.equal(model.lm_head.online_value, torch.tensor([7.0]))
assert torch.equal(model.lm_head.traversal.weight, torch.tensor([[3.0]]))
assert model.lm_head._self_correction_bank.grounded_trigger_pressure[1] == 1.0
bank.validate_pending_observation(selected_observation)
assert torch.equal(
cast(
torch.Tensor,
bank.pending_prompt_fingerprint([selected_observation]),
),
torch.arange(32, dtype=torch.uint8),
)
with pytest.raises(ValueError, match="do not match"):
bank.pending_correction_state(
[{**selected_observation, "args": {"path": "changed.txt"}}]
)
bank.checkpoint_active()
session_path = next((state_root / "sessions").glob("*/000001.safetensors"))
with _SAFE_OPEN(str(session_path), framework="pt", device="cpu") as handle:
assert all(key.isdigit() and len(key) == 6 for key in handle.keys())
session_metadata = handle.metadata() or {}
assert session_metadata["000001"] == "12"
assert all(key.isdigit() and len(key) == 6 for key in session_metadata)
reloaded = Model()
restored = NexumSessionStateBank(reloaded, device="cpu", state_root=state_root)
restored.activate("session-a")
assert torch.equal(reloaded.lm_head.online_value, torch.tensor([7.0]))
assert reloaded.lm_head._self_correction_bank.grounded_trigger_pressure[1] == 1.0
restored.validate_pending_observation(selected_observation)
assert torch.equal(
cast(
torch.Tensor,
restored.pending_prompt_fingerprint([selected_observation]),
),
torch.arange(32, dtype=torch.uint8),
)
pending_trigger, pending_path = restored.pending_correction_state(
[selected_observation]
)
assert torch.equal(cast(torch.Tensor, pending_trigger), torch.tensor([0.75, 0.25]))
assert len(pending_path) == 1
assert pending_path[0][0] == 1
assert torch.equal(pending_path[0][1], torch.tensor([0, 2]))
pending_none = restored.pending_none_outcome_state([selected_observation])
assert pending_none is not None
assert torch.equal(
pending_none.route_pair_ids_t,
torch.tensor([[[0, 1]], [[2, 3]]]),
)
assert torch.equal(
pending_none.route_weight_t,
torch.tensor([[[0.75, 0.25]], [[0.4, 0.6]]]),
)
assert torch.equal(
pending_none.path_acceptance_t,
torch.tensor([[True], [False]]),
)
assert pending_none.adaptation_t == 0.5
failed_action = {
"name": "Read",
"args": {"path": "missing.txt"},
"ok": False,
"executed": True,
}
corrected_failure = {
**failed_action,
"args": {"path": "replacement.txt"},
}
assert torch.equal(
restored.self_improvement_required_signal([failed_action]),
torch.tensor(False),
)
assert torch.equal(
restored.repeated_failure_signal([failed_action]), torch.tensor(False)
)
restored.commit_observations([failed_action])
assert torch.equal(
restored.self_improvement_required_signal([corrected_failure]),
torch.tensor(True),
)
assert torch.equal(
restored.repeated_failure_signal([failed_action]), torch.tensor(True)
)
successful_observation = {
"name": "Read",
"args": {"path": "verified.txt"},
"ok": True,
"executed": True,
"output": "verified\n",
}
assert torch.equal(
restored.repeated_success_signal([successful_observation]),
torch.tensor(False),
)
restored.commit_observations([successful_observation])
assert torch.equal(
restored.self_improvement_required_signal([corrected_failure]),
torch.tensor(False),
)
assert torch.equal(
restored.repeated_success_signal([successful_observation]),
torch.tensor(True),
)
restored.commit_observations([successful_observation])
assert torch.equal(
restored.self_improvement_required_signal([successful_observation]),
torch.tensor(True),
)
assert torch.equal(
restored.repeated_success_signal(
[{**successful_observation, "output": "updated\n"}]
),
torch.tensor(False),
)
assert restored.observation_nonce_consumed("nonce-a") is False
restored.commit_observation_nonces(("nonce-a",))
assert restored.observation_nonce_consumed("nonce-a") is True
restored.checkpoint_active()
cold_model = Model()
cold = NexumSessionStateBank(cold_model, device="cpu", state_root=state_root)
cold.activate("session-a")
assert cold.observation_nonce_consumed("nonce-a") is True
assert torch.equal(
cold.repeated_failure_signal([failed_action]), torch.tensor(True)
)
assert torch.equal(
cold.self_improvement_required_signal([corrected_failure]),
torch.tensor(True),
)
assert torch.equal(
cold.repeated_success_signal([successful_observation]), torch.tensor(True)
)
cold.commit_observations([{**failed_action, "ok": True}])
assert torch.equal(
cold.repeated_failure_signal([failed_action]), torch.tensor(False)
)
assert torch.allclose(pending_path[0][2], torch.tensor([0.25, 0.75]))
restored.commit_observations()
def test_grounded_learning_updates_only_future_sessions_and_cold_restores(
tmp_path: Path,
) -> None:
class PagedState(torch.nn.Module):
session_marker: torch.Tensor
def __init__(self) -> None:
super().__init__()
self.register_buffer("session_marker", torch.zeros(4, dtype=torch.long))
self.source_t = torch.tensor([0, 1], dtype=torch.long)
self.donor_t = torch.tensor([0, 1], dtype=torch.long)
self.consumed_t = torch.zeros((), dtype=torch.float32)
def begin_session(self, session_id_t: torch.Tensor) -> None:
self.session_marker.copy_(session_id_t)
@staticmethod
def base_topology_boundary() -> tuple[torch.Tensor, torch.Tensor, torch.Tensor]:
base_t = torch.tensor([0, 1], dtype=torch.long)
return base_t, base_t.clone(), torch.zeros(())
def shared_topology_boundary(
self,
) -> tuple[torch.Tensor, torch.Tensor, torch.Tensor]:
return (
self.source_t.clone(),
self.donor_t.clone(),
self.consumed_t.clone(),
)
def configure_shared_topology_boundary(
self,
source_t: torch.Tensor,
donor_t: torch.Tensor,
consumed_t: torch.Tensor,
) -> None:
self.source_t = source_t.detach().cpu().clone()
self.donor_t = donor_t.detach().cpu().clone()
self.consumed_t = consumed_t.detach().cpu().reshape(()).clone()
class CorrectionState(torch.nn.Module):
trigger_success: torch.Tensor
trigger_failure: torch.Tensor
def __init__(self) -> None:
super().__init__()
self.register_buffer("trigger_success", torch.zeros(3))
self.register_buffer("trigger_failure", torch.zeros(3))
class PathState(torch.nn.Module):
pro_score: torch.Tensor
anti_score: torch.Tensor
expert_identity_strength: torch.Tensor
layer_confidence: torch.Tensor
domain_cell_pro: torch.Tensor
domain_cell_anti: torch.Tensor
domain_growth_pressure: torch.Tensor
def __init__(self) -> None:
super().__init__()
self.register_buffer("pro_score", torch.zeros(2, 3))
self.register_buffer("anti_score", torch.zeros(2, 3))
self.register_buffer("expert_identity_strength", torch.zeros(2, 3))
self.register_buffer("layer_confidence", torch.zeros(2))
self.register_buffer("domain_cell_pro", torch.zeros(2, 2))
self.register_buffer("domain_cell_anti", torch.zeros(2, 2))
self.register_buffer("domain_growth_pressure", torch.zeros(2, 2))
def clear_self_correction_bias(self) -> None:
return
def clear_merkle_route(self) -> None:
return
class Head(torch.nn.Module):
private_value: torch.Tensor
def __init__(self) -> None:
super().__init__()
self.register_buffer("private_value", torch.zeros(1))
self.traversal = torch.nn.Linear(1, 1, bias=False)
self.gate = torch.nn.Linear(1, 1, bias=False)
self._self_correction_bank = CorrectionState()
self._path_registry = PathState()
self._none_paged_stack = PagedState()
self.experts = torch.nn.ModuleList()
class Model(torch.nn.Module):
def __init__(self) -> None:
super().__init__()
self.lm_head = Head()
state_root = tmp_path / "state"
model = Model()
bank = NexumSessionStateBank(model, device="cpu", state_root=state_root)
assert bank.self_improvement_ready is True
bank.activate("existing-session")
bank.activate("learning-session")
before = bank.capture_grounded_learning_state()
with torch.no_grad():
model.lm_head.private_value.fill_(9.0)
model.lm_head._self_correction_bank.trigger_failure[1].add_(1.0)
model.lm_head._path_registry.anti_score[0, 1].add_(1.0)
model.lm_head._path_registry.expert_identity_strength[0, 1].sub_(0.25)
model.lm_head._path_registry.domain_cell_anti[1, 0].add_(1.0)
model.lm_head._path_registry.domain_growth_pressure[1, 0].add_(1.0)
model.lm_head._none_paged_stack.source_t = torch.tensor([0, 1, 1])
model.lm_head._none_paged_stack.donor_t = torch.tensor([0, 1, 1])
model.lm_head._none_paged_stack.consumed_t = torch.tensor(1.0)
retained_evidence = torch.arange(24, dtype=torch.float32).reshape(4, 6)
promoted = bank.promote_grounded_observation(
"a" * 24,
before,
evidence_token_t=retained_evidence,
evidence_context_length_t=torch.tensor([2]),
)
assert promoted["learning_applied"] is True
assert promoted["learning_resumed"] is False
assert promoted["future_session_prior_updated"] is True
assert promoted["learning_slots_updated"] == 5
assert promoted["learning_evidence_persisted"] is True
assert promoted["retained_learning_evidence_tokens"] == 4
assert promoted["retained_learning_success_evidence_tokens"] == 0
assert promoted["retained_learning_failure_evidence_tokens"] == 4
assert promoted["shared_topology_bank_count"] == 3
assert bank.shared_learning_evidence_token().numel() == 0
assert torch.equal(
bank.shared_learning_anti_evidence_token(),
retained_evidence,
)
promoted_generation = promoted["learning_generation"]
replayed = bank.promote_grounded_observation("a" * 24, before)
assert replayed["learning_applied"] is True
assert replayed["learning_resumed"] is True
assert replayed["learning_generation"] == promoted_generation
neutral_before = bank.capture_grounded_learning_state()
neutral = bank.promote_grounded_observation("b" * 24, neutral_before)
assert neutral["learning_applied"] is False
assert neutral["learning_generation"] == promoted_generation
bank.activate("untrusted-session")
untrusted_before = bank.capture_grounded_learning_state()
with torch.no_grad():
model.lm_head._self_correction_bank.trigger_failure[2].add_(1.0)
rejected = bank.promote_grounded_observation(
"c" * 24,
untrusted_before,
validation={
"eligible": False,
"reason": "untrusted_execution",
"evidence_sha256": "d" * 64,
"observation_count": 1,
"trusted_observation_count": 0,
"untrusted_observation_count": 1,
"tool_count": 1,
"output_bytes": 32,
"elapsed_s": 0.25,
},
)
assert rejected["learning_applied"] is False
assert rejected["learning_candidate_validated"] is False
assert rejected["learning_rejection_reason"] == "untrusted_execution"
assert rejected["learning_generation"] == promoted_generation
bank.activate("existing-session")
assert torch.equal(
model.lm_head._self_correction_bank.trigger_failure, torch.zeros(3)
)
assert torch.equal(model.lm_head._path_registry.anti_score, torch.zeros(2, 3))
bank.activate("future-session")
assert torch.equal(model.lm_head.private_value, torch.zeros(1))
assert model.lm_head._self_correction_bank.trigger_failure[1] == 1.0
assert model.lm_head._path_registry.anti_score[0, 1] == 1.0
assert model.lm_head._path_registry.expert_identity_strength[0, 1] == -0.25
assert model.lm_head._path_registry.domain_cell_anti[1, 0] == 1.0
assert model.lm_head._path_registry.domain_growth_pressure[1, 0] == 1.0
bank.activate("learning-session")
assert torch.equal(model.lm_head.private_value, torch.tensor([9.0]))
successful_evidence = torch.arange(12, dtype=torch.float32).reshape(2, 6) + 100.0
bank.mark_self_improvement_pending()
pending_model = Model()
pending_reload = NexumSessionStateBank(
pending_model,
device="cpu",
state_root=state_root,
)
pending_reload.activate("learning-session")
assert bool(
pending_reload.self_improvement_success_signal(
[{"name": "Read", "args": {}, "ok": True, "executed": True}]
)
)
assert not bool(bank.self_improvement_success_signal([]))
assert not bool(
bank.self_improvement_success_signal(
[
{"name": "Read", "args": {}, "ok": True, "executed": True},
{"name": "Write", "args": {}, "ok": False, "executed": True},
]
)
)
assert bool(
bank.self_improvement_success_signal(
[
{"name": "Read", "args": {}, "ok": True, "executed": True},
{"name": "Write", "args": {}, "ok": True, "executed": True},
]
)
)
consolidated = bank.consolidate_grounded_success(
"d" * 24,
evidence_token_t=successful_evidence,
evidence_context_length_t=torch.tensor([1]),
)
assert consolidated["learning_applied"] is True
assert consolidated["learning_slots_updated"] == 0
assert consolidated["future_session_prior_updated"] is True
assert consolidated["retained_learning_evidence_tokens"] == 6
assert consolidated["retained_learning_success_evidence_tokens"] == 2
assert consolidated["retained_learning_failure_evidence_tokens"] == 4
assert not bool(
bank.self_improvement_success_signal(
[{"name": "Read", "args": {}, "ok": True, "executed": True}]
)
)
assert torch.equal(
bank.shared_learning_evidence_token(),
successful_evidence,
)
assert torch.equal(
bank.shared_learning_anti_evidence_token(),
retained_evidence,
)
shared_evidence = bank.shared_learning_evidence_packet()
assert torch.equal(
shared_evidence.evidence_episode_length_t,
torch.tensor([2]),
)
assert torch.equal(
shared_evidence.evidence_context_length_t,
torch.tensor([1]),
)
assert torch.equal(
shared_evidence.anti_evidence_episode_length_t,
torch.tensor([4]),
)
assert torch.equal(
shared_evidence.anti_evidence_context_length_t,
torch.tensor([2]),
)
generation = consolidated["learning_generation"]
assert generation == promoted_generation + 1
bank.checkpoint_active()
learning_path = state_root / "learning" / "000001.safetensors"
with _SAFE_OPEN(str(learning_path), framework="pt", device="cpu") as handle:
assert all(key.isdigit() and len(key) == 6 for key in handle.keys())
metadata = handle.metadata() or {}
assert metadata["000001"] == "1"
assert all(key.isdigit() and len(key) == 6 for key in metadata)
assert "learning-session" not in (
state_root / "learning" / "000002.jsonl"
).read_text(encoding="utf-8")
evidence_path = state_root / "learning" / "000005" / "000001.safetensors"
with _SAFE_OPEN(str(evidence_path), framework="pt", device="cpu") as handle:
assert tuple(handle.keys()) == tuple(f"{index:06d}" for index in range(1, 14))
assert handle.get_tensor("000001").numel() == 0
assert torch.equal(handle.get_tensor("000002"), retained_evidence)
assert handle.get_tensor("000003").numel() == 0
assert handle.get_tensor("000004").numel() == 0
assert torch.equal(handle.get_tensor("000005"), torch.tensor([4]))
assert torch.equal(handle.get_tensor("000006"), torch.tensor([2]))
assert torch.equal(handle.get_tensor("000007"), torch.tensor([0, 1, 1]))
assert torch.equal(handle.get_tensor("000008"), torch.tensor([0, 1, 1]))
assert torch.equal(handle.get_tensor("000009"), torch.tensor([1.0]))
assert handle.get_tensor("000010").numel() == 0
assert handle.get_tensor("000011").numel() == 0
assert handle.get_tensor("000012").numel() == 0
assert torch.equal(handle.get_tensor("000013"), torch.tensor([0]))
evidence_metadata = handle.metadata() or {}
assert evidence_metadata["000001"] == "6"
assert all(key.isdigit() and len(key) == 6 for key in evidence_metadata)
consolidated_evidence_path = (
state_root / "learning" / "000005" / "000002.safetensors"
)
with _SAFE_OPEN(
str(consolidated_evidence_path), framework="pt", device="cpu"
) as handle:
assert torch.equal(handle.get_tensor("000001"), successful_evidence)
assert torch.equal(handle.get_tensor("000002"), retained_evidence)
assert torch.equal(handle.get_tensor("000003"), torch.tensor([2]))
assert torch.equal(handle.get_tensor("000004"), torch.tensor([1]))
assert torch.equal(handle.get_tensor("000005"), torch.tensor([4]))
assert torch.equal(handle.get_tensor("000006"), torch.tensor([2]))
assert handle.get_tensor("000010").numel() == 0
assert torch.equal(handle.get_tensor("000011"), torch.tensor([0]))
assert handle.get_tensor("000012").numel() == 0
assert torch.equal(handle.get_tensor("000013"), torch.tensor([0]))
reloaded = Model()
restored = NexumSessionStateBank(reloaded, device="cpu", state_root=state_root)
assert restored.learning_generation == generation
assert torch.equal(
restored.shared_learning_evidence_token(),
successful_evidence,
)
assert torch.equal(
restored.shared_learning_anti_evidence_token(),
retained_evidence,
)
restored_evidence = restored.shared_learning_evidence_packet()
assert torch.equal(
restored_evidence.evidence_episode_length_t,
torch.tensor([2]),
)
assert torch.equal(
restored_evidence.evidence_context_length_t,
torch.tensor([1]),
)
assert torch.equal(
restored_evidence.anti_evidence_episode_length_t,
torch.tensor([4]),
)
assert torch.equal(
restored_evidence.anti_evidence_context_length_t,
torch.tensor([2]),
)
assert restored.shared_topology_bank_count == 3
assert torch.equal(
reloaded.lm_head._none_paged_stack.source_t,
torch.tensor([0, 1, 1]),
)
restored.activate("cold-future-session")
assert torch.equal(reloaded.lm_head.private_value, torch.zeros(1))
assert reloaded.lm_head._self_correction_bank.trigger_failure[1] == 1.0
assert reloaded.lm_head._path_registry.anti_score[0, 1] == 1.0
status = restored.learning_status()
assert status["decision_chain_sha256"]
assert status["rollback_generations"] == [0, 1]
assert status["decision_counts"] == {
"promotion_applied": 2,
"promotion_rejected": 2,
"rollback_applied": 0,
}
rollback = restored.rollback_learning(
target_generation=0,
expected_generation=generation,
)
assert rollback["source_generation"] == 0
assert rollback["previous_generation"] == generation
assert rollback["learning_generation"] == generation + 1
assert restored.shared_learning_evidence_token().numel() == 0
assert restored.shared_learning_anti_evidence_token().numel() == 0
assert restored.shared_topology_bank_count == 2
restored.activate("post-rollback-session")
assert torch.equal(
reloaded.lm_head._self_correction_bank.trigger_failure,
torch.zeros(3),
)
assert torch.equal(
reloaded.lm_head._path_registry.anti_score,
torch.zeros(2, 3),
)
cold_model = Model()
cold = NexumSessionStateBank(cold_model, device="cpu", state_root=state_root)
assert cold.learning_generation == generation + 1
assert cold.learning_status()["decision_counts"]["rollback_applied"] == 1
assert cold.shared_learning_evidence_token().numel() == 0
assert cold.shared_learning_anti_evidence_token().numel() == 0
assert cold.shared_topology_bank_count == 2
cold.activate("cold-post-rollback-session")
assert torch.equal(
cold_model.lm_head._self_correction_bank.trigger_failure,
torch.zeros(3),
)
decision_path = state_root / "learning" / "000002.jsonl"
decisions = [
json.loads(line)
for line in decision_path.read_text(encoding="utf-8").splitlines()
if line.strip()
]
previous_sha256 = ""
for decision in decisions:
assert decision["previous_sha256"] == previous_sha256
supplied_sha256 = decision.pop("receipt_sha256")
canonical = json.dumps(
decision,
sort_keys=True,
separators=(",", ":"),
).encode("utf-8")
assert supplied_sha256 == hashlib.sha256(canonical).hexdigest()
previous_sha256 = supplied_sha256
def test_core_api_forwards_grounded_semantic_evidence_tensor() -> None:
evidence_token_t = torch.randn(5, 7)
with mock.patch("nexum_core.api._observe_tool_outcomes") as observe:
observe.return_value = object()
observe_tool_outcomes_api(
object(),
[],
task_id="semantic-api-boundary",
config=object(),
grounded_evidence_token_t=evidence_token_t,
)
assert observe.call_args.kwargs["grounded_evidence_token_t"] is evidence_token_t
def test_execution_outcomes_update_correction_without_positive_task_route_credit(
tmp_path: Path,
) -> None:
cfg = NexumConfig.tiny()
cfg.execution_grounding_ledger_path = str(tmp_path / "execution.jsonl")
class Registry:
def __init__(self) -> None:
self.pro_score = torch.zeros(2, 3)
self.active_domain = torch.tensor(0)
self.active_subdomain = torch.tensor(0)
self.route_updates = 0
self.domain_updates = 0
def reinforce_traversal_path(self, *_args: object, **kwargs: object) -> None:
assert (tmp_path / "execution.jsonl").is_file()
assert float(cast(torch.Tensor, kwargs["positive_signal"])) == 0.0
self.route_updates += 1
def record_domain_outcome(self, *_args: object, **kwargs: object) -> None:
assert cast(torch.Tensor, kwargs["success"]).item() is False
self.domain_updates += 1
class Head(torch.nn.Module):
def __init__(self) -> None:
super().__init__()
self.anchor = torch.nn.Parameter(torch.zeros(()))
object.__setattr__(
self, "_self_correction_bank", SelfCorrectionTriggerBank(cfg)
)
object.__setattr__(self, "_path_registry", Registry())
def self_correction_packet(self) -> None:
return None
owner = SimpleNamespace(lm_head=Head())
bank = owner.lm_head.__dict__["_self_correction_bank"]
trigger_mass = torch.ones(bank.trigger_slots)
before_success = bank.trigger_success.clone()
before_failure = bank.trigger_failure.clone()
registry = owner.lm_head.__dict__["_path_registry"]
traversal_path = [(0, torch.tensor([1]), torch.tensor([1.0]))]
success_receipt = observe_tool_outcomes(
owner,
[
{
"name": "Bash",
"args": {"command": "true"},
"ok": True,
"output": "",
"executed": True,
"exit_code": 0,
}
],
task_id="task",
config=cfg,
correction_trigger_mass=trigger_mass,
correction_traversal_path=traversal_path,
)
assert success_receipt.verification_updated is True
assert success_receipt.correction_updated is False
assert success_receipt.self_correction_applied is False
assert success_receipt.self_improvement_required is False
assert success_receipt.self_improvement_applied is False
assert success_receipt.route_updated is False
assert success_receipt.domain_updated is False
assert torch.all(bank.trigger_success > before_success)
assert torch.equal(bank.trigger_failure, before_failure)
assert registry.route_updates == 0
assert registry.domain_updates == 0
failure_receipt = observe_tool_outcomes(
owner,
[
{
"name": "Bash",
"args": {"command": "false"},
"ok": False,
"output": "",
"executed": True,
"exit_code": 1,
}
],
task_id="task",
config=cfg,
correction_trigger_mass=trigger_mass,
correction_traversal_path=traversal_path,
)
assert failure_receipt.verification_updated is False
assert failure_receipt.correction_updated is True
assert failure_receipt.self_correction_applied is True
assert failure_receipt.self_improvement_required is False
assert failure_receipt.self_improvement_applied is False
assert failure_receipt.route_updated is False
assert failure_receipt.domain_updated is False
assert torch.all(bank.trigger_success > before_success)
failure_delta = bank.trigger_failure - before_failure
assert failure_delta[1] == 1.0
assert torch.count_nonzero(failure_delta) == 1
assert registry.route_updates == 0
assert registry.domain_updates == 0
improvement_receipt = observe_tool_outcomes(
owner,
[
{
"name": "Bash",
"args": {"command": "exit 2"},
"ok": False,
"output": "",
"executed": True,
"exit_code": 2,
}
],
task_id="task",
config=cfg,
correction_trigger_mass=trigger_mass,
correction_traversal_path=traversal_path,
improvement_required_t=torch.tensor(True),
)
assert improvement_receipt.self_correction_applied is True
assert improvement_receipt.self_improvement_required is True
assert improvement_receipt.self_improvement_applied is True
assert improvement_receipt.route_updated is True
assert improvement_receipt.domain_updated is True
assert registry.route_updates == 1
assert registry.domain_updates == 1
rejected_receipt = observe_tool_outcomes(
owner,
[
{
"name": "Bash",
"args": {},
"ok": False,
"output": "invalid arguments",
"executed": False,
"exit_code": None,
}
],
task_id="task",
config=cfg,
correction_trigger_mass=trigger_mass,
correction_traversal_path=traversal_path,
)
assert rejected_receipt.failures == 1
assert rejected_receipt.self_correction_applied is True
assert rejected_receipt.self_improvement_required is False
assert rejected_receipt.self_improvement_applied is False
assert rejected_receipt.route_updated is False
assert registry.route_updates == 1
assert registry.domain_updates == 1
before_targeted = bank.trigger_failure.clone()
repeated_receipt = observe_tool_outcomes_api(
owner,
[
{
"name": "Read",
"args": {"path": "missing.txt"},
"ok": False,
"output": "",
"executed": True,
"exit_code": None,
}
],
task_id="task",
config=cfg,
correction_trigger_mass=torch.zeros_like(trigger_mass),
correction_traversal_path=traversal_path,
repeated_action_t=torch.tensor(True),
)
targeted_delta = bank.trigger_failure - before_targeted
assert repeated_receipt.repeated_action_detected is True
assert repeated_receipt.repeated_success_detected is False
assert repeated_receipt.self_correction_applied is True
assert repeated_receipt.self_improvement_required is True
assert repeated_receipt.self_improvement_applied is True
assert repeated_receipt.route_updated is True
assert repeated_receipt.domain_updated is True
assert targeted_delta[2] == 1.0
assert targeted_delta[5] == 1.0
assert torch.count_nonzero(targeted_delta) == 2
before_stagnation = bank.trigger_failure.clone()
before_stagnation_success = bank.trigger_success.clone()
repeated_success_receipt = observe_tool_outcomes_api(
owner,
[
{
"name": "Read",
"args": {"path": "verified.txt"},
"ok": True,
"output": "verified\n",
"executed": True,
"exit_code": None,
}
],
task_id="task",
config=cfg,
correction_trigger_mass=torch.ones_like(trigger_mass),
correction_traversal_path=traversal_path,
repeated_success_t=torch.tensor(True),
)
stagnation_delta = bank.trigger_failure - before_stagnation
stagnation_success_delta = bank.trigger_success - before_stagnation_success
assert repeated_success_receipt.successes == 1
assert repeated_success_receipt.failures == 0
assert repeated_success_receipt.verification_updated is True
assert repeated_success_receipt.route_updated is False
assert repeated_success_receipt.domain_updated is False
assert repeated_success_receipt.self_correction_applied is True
assert repeated_success_receipt.self_improvement_required is False
assert repeated_success_receipt.self_improvement_applied is False
assert repeated_success_receipt.repeated_action_detected is True
assert repeated_success_receipt.repeated_success_detected is True
assert stagnation_delta[5] == 1.0
assert torch.count_nonzero(stagnation_delta) == 1
assert stagnation_success_delta[5] == 0.0
assert torch.count_nonzero(stagnation_success_delta) == trigger_mass.numel() - 1
improved_stagnation_receipt = observe_tool_outcomes_api(
owner,
[
{
"name": "Read",
"args": {"path": "verified.txt"},
"ok": True,
"output": "verified\n",
"executed": True,
"exit_code": 0,
}
],
task_id="task",
config=cfg,
correction_trigger_mass=torch.zeros_like(trigger_mass),
correction_traversal_path=traversal_path,
repeated_success_t=torch.tensor(True),
improvement_required_t=torch.tensor(True),
)
assert improved_stagnation_receipt.successes == 1
assert improved_stagnation_receipt.failures == 0
assert improved_stagnation_receipt.self_improvement_required is True
assert improved_stagnation_receipt.self_improvement_applied is True
assert improved_stagnation_receipt.route_updated is True
assert improved_stagnation_receipt.domain_updated is True
def test_execution_outcome_reaches_exact_model_selected_none_route(
tmp_path: Path,
) -> None:
cfg = NexumConfig.tiny()
cfg.execution_grounding_ledger_path = str(tmp_path / "execution.jsonl")
observed: list[tuple[torch.Tensor, ...]] = []
growth_calls: list[tuple[torch.Tensor, torch.Tensor]] = []
class Registry:
active_domain = torch.tensor(0)
active_subdomain = torch.tensor(0)
@staticmethod
def record_domain_outcome(*_args: object, **_kwargs: object) -> None:
return
@staticmethod
def domain_growth_score(
_domain_idx: torch.Tensor,
_subdomain_idx: torch.Tensor,
) -> torch.Tensor:
return torch.tensor(2.25)
class Stack:
def observe_grounded_outcome(
self,
route_ids_t: torch.Tensor,
route_weight_t: torch.Tensor,
acceptance_t: torch.Tensor,
outcome_t: torch.Tensor,
verification_t: torch.Tensor,
adaptation_t: torch.Tensor,
grounded_evidence_token_t: torch.Tensor | None = None,
) -> object:
assert (tmp_path / "execution.jsonl").is_file()
observed.append(
(
route_ids_t.clone(),
route_weight_t.clone(),
acceptance_t.clone(),
outcome_t.clone(),
verification_t.clone(),
adaptation_t.clone(),
(
grounded_evidence_token_t.clone()
if grounded_evidence_token_t is not None
else torch.empty(0)
),
)
)
return SimpleNamespace(
generation_t=torch.tensor(3),
routed_authority_count_t=torch.tensor(4),
update_mass_t=torch.tensor(0.125),
semantic_evidence_applied_t=torch.tensor(
grounded_evidence_token_t is not None
),
semantic_evidence_token_count_t=torch.tensor(
(
grounded_evidence_token_t.shape[0]
if grounded_evidence_token_t is not None
else 0
)
),
)
def grow_from_model_pressure(
self,
donor_bank_idx_t: torch.Tensor,
model_pressure_t: torch.Tensor,
) -> object:
growth_calls.append((donor_bank_idx_t.clone(), model_pressure_t.clone()))
return SimpleNamespace(
banks_added_t=torch.tensor(2),
bank_count_t=torch.tensor(4),
model_pressure_t=model_pressure_t,
)
class Head(torch.nn.Module):
def __init__(self) -> None:
super().__init__()
self.anchor = torch.nn.Parameter(torch.zeros(()))
self.stack = Stack()
object.__setattr__(self, "_path_registry", Registry())
def none_paged_stack(self) -> Stack:
return self.stack
route_ids_t = torch.tensor([[[0, 1]], [[2, 3]]])
route_weight_t = torch.tensor([[[0.6, 0.4]], [[0.25, 0.75]]])
acceptance_t = torch.tensor([[True], [False]])
adaptation_t = torch.tensor(0.2)
evidence_token_t = torch.randn(7, cfg.token_dim)
receipt = observe_tool_outcomes(
SimpleNamespace(lm_head=Head()),
[
{
"name": "Read",
"args": {"path": "missing"},
"ok": False,
"output": "not found",
"executed": True,
"exit_code": 1,
}
],
task_id="grounded-page-task",
config=cfg,
improvement_required_t=torch.tensor(True),
none_route_pair_ids_t=route_ids_t,
none_route_weight_t=route_weight_t,
none_path_acceptance_t=acceptance_t,
none_adaptation_t=adaptation_t,
grounded_evidence_token_t=evidence_token_t,
)
assert receipt.none_paged_updated is True
assert receipt.self_improvement_required is True
assert receipt.self_improvement_applied is True
assert receipt.none_paged_generation == 3
assert receipt.none_paged_authorities_updated == 4
assert receipt.none_paged_update_mass == pytest.approx(0.125)
assert receipt.none_semantic_evidence_applied is True
assert receipt.none_semantic_evidence_tokens == 7
assert receipt.none_banks_grown == 2
assert receipt.none_bank_count == 4
assert receipt.none_growth_pressure == pytest.approx(2.25)
assert len(growth_calls) == 1
assert growth_calls[0][0] == 1
assert growth_calls[0][1] == pytest.approx(torch.tensor(2.25))
assert len(observed) == 1
assert torch.equal(observed[0][0], route_ids_t)
assert torch.equal(observed[0][1], route_weight_t)
assert torch.equal(observed[0][2], acceptance_t)
assert observed[0][3] == 0
assert observed[0][4] == 1
assert observed[0][5] == adaptation_t
assert torch.equal(observed[0][6], evidence_token_t)
def test_grounded_failure_raises_next_turn_correction_pressure() -> None:
cfg = NexumConfig.tiny()
bank = SelfCorrectionTriggerBank(cfg)
hidden = torch.zeros(1, 1, cfg.hidden_size)
with torch.no_grad():
bank.token_to_hidden.weight.fill_(0.01)
before = bank(hidden)
focused_mass = torch.zeros_like(before.trigger_mass)
focused_mass[1] = 1.0
bank.record_outcome_mass(
focused_mass,
success=torch.tensor(False),
magnitude=torch.tensor(1.0),
)
after = bank(hidden)
assert after.trigger_probs[..., 1].mean() > before.trigger_probs[..., 1].mean()
assert bank.grounded_trigger_pressure[1] == 1.0
assert after.confidence.mean() > before.confidence.mean()
assert torch.linalg.vector_norm(after.corrected_hidden) > torch.linalg.vector_norm(
before.corrected_hidden
)
assert not torch.equal(after.axis_a_bias, before.axis_a_bias)
bank.record_outcome_mass(
focused_mass,
success=torch.tensor(True),
magnitude=torch.tensor(1.0),
)
assert bank.grounded_trigger_pressure[1] == 0.0
def test_correction_confidence_modulates_current_recursive_route() -> None:
cfg = NexumConfig.tiny()
head = CapacityLMHead(
torch.nn.Linear(cfg.hidden_size, cfg.vocab_size, bias=False),
cfg,
n_layers=1,
)
hidden = torch.randn(1, 1, cfg.hidden_size)
route_delta = torch.randn_like(hidden)
token_bridge = NexumTBR(cfg)
with (
mock.patch.object(head, "expert_delta_rbo", return_value=route_delta),
mock.patch.object(
head,
"self_correction_packet",
return_value=SimpleNamespace(confidence=torch.zeros(1, 1, 1)),
),
torch.no_grad(),
):
base_logits = head.orig(hidden)
gated_logits = head.forward_generation_rbo(hidden, token_bridge)
assert torch.equal(gated_logits, base_logits)
assert head._last_delta is not None
assert torch.count_nonzero(head._last_delta) == 0
with (
mock.patch.object(head, "expert_delta_rbo", return_value=route_delta),
mock.patch.object(
head,
"self_correction_packet",
return_value=SimpleNamespace(confidence=torch.ones(1, 1, 1)),
),
torch.no_grad(),
):
active_logits = head.forward_generation_rbo(hidden, token_bridge)
assert not torch.equal(active_logits, base_logits)
assert torch.allclose(
torch.sigmoid(head.native_decode_confidence.route_strength_logit),
torch.tensor(cfg.capacity_delta_rms_ratio),
)
assert not hasattr(head, "_last_rbo_route_delta")
assert not hasattr(head, "_last_rbo_route_confidence_t")
def test_context_recursive_route_preserves_training_graph_without_route_cache() -> None:
cfg = NexumConfig.tiny()
head = CapacityLMHead(
torch.nn.Linear(cfg.hidden_size, cfg.vocab_size, bias=False),
cfg,
n_layers=1,
)
hidden = torch.randn(2, 4, cfg.hidden_size)
context_hidden = torch.randn(2, 3, cfg.hidden_size)
route_delta = torch.randn(2, 1, cfg.hidden_size, requires_grad=True)
token_bridge = NexumTBR(cfg)
observed: list[torch.Tensor] = []
def context_delta(value: torch.Tensor, bridge: object) -> torch.Tensor:
assert bridge is token_bridge
observed.append(value.detach().clone())
return route_delta
with (
mock.patch.object(head, "expert_delta_rbo", side_effect=context_delta),
mock.patch.object(
head,
"self_correction_packet",
return_value=SimpleNamespace(confidence=torch.ones(2, 1, 1)),
),
):
logits = head.forward_context_rbo(hidden, context_hidden, token_bridge)
torch.autograd.backward(logits.sum())
assert len(observed) == 1
assert torch.equal(observed[0], context_hidden[:, -1:, :])
assert head._last_delta is not None
assert head._last_delta.shape == hidden.shape
assert route_delta.grad is not None
assert torch.count_nonzero(route_delta.grad) > 0
assert not hasattr(head, "_last_rbo_route_delta")
assert not hasattr(head, "_last_rbo_route_confidence_t")
def test_context_recursive_route_runs_real_batched_rbo() -> None:
cfg = NexumConfig.tiny()
head = CapacityLMHead(
torch.nn.Linear(cfg.hidden_size, cfg.vocab_size, bias=False),
cfg,
n_layers=1,
)
token_bridge = NexumTBR(cfg)
hidden = torch.randn(2, 4, cfg.hidden_size)
context_hidden = torch.randn(2, 3, cfg.hidden_size)
logits = head.forward_context_rbo(hidden, context_hidden, token_bridge)
assert logits.shape == (2, 4, cfg.vocab_size)
assert head._last_delta is not None
assert head._last_delta.shape == hidden.shape
def test_recursive_generation_route_refreshes_after_request_context() -> None:
cfg = NexumConfig.tiny()
head = CapacityLMHead(
torch.nn.Linear(cfg.hidden_size, cfg.vocab_size, bias=False),
cfg,
n_layers=1,
)
token_bridge = NexumTBR(cfg)
object.__setattr__(head, "_nexum_token_bridge", token_bridge)
first_hidden = torch.randn(1, 1, cfg.hidden_size)
second_hidden = torch.randn(1, 1, cfg.hidden_size)
observed: list[torch.Tensor] = []
def live_delta(value: torch.Tensor, bridge: object) -> torch.Tensor:
assert bridge is token_bridge
observed.append(value.detach().clone())
return torch.zeros_like(value)
with (
mock.patch.object(
head,
"expert_delta_rbo",
side_effect=live_delta,
),
mock.patch.object(
head,
"self_correction_packet",
return_value=SimpleNamespace(confidence=torch.ones(1, 1, 1)),
),
torch.no_grad(),
):
first_logits = head(first_hidden)
second_logits = head(second_hidden)
assert not torch.equal(first_logits, second_logits)
assert len(observed) == 2
assert torch.equal(observed[0], first_hidden)
assert torch.equal(observed[1], second_hidden)
assert int(head._rbo_generation_forward_count_t) == 2
def test_request_context_prepares_generation_route_cache() -> None:
cfg = NexumConfig.tiny()
head = CapacityLMHead(
torch.nn.Linear(cfg.hidden_size, cfg.vocab_size, bias=False),
cfg,
n_layers=1,
)
token_bridge = NexumTBR(cfg)
object.__setattr__(head, "_nexum_token_bridge", token_bridge)
observed: list[torch.Tensor] = []
def request_delta(value: torch.Tensor, bridge: object) -> torch.Tensor:
assert bridge is token_bridge
observed.append(value.detach().clone())
return torch.ones_like(value) * 0.125
with (
mock.patch.object(head, "expert_delta_rbo", side_effect=request_delta),
mock.patch.object(
head,
"self_correction_packet",
return_value=SimpleNamespace(confidence=torch.ones(1, 1, 1)),
),
):
head.initialize_request_context(torch.tensor([[1, 2, 3]], dtype=torch.long))
assert len(observed) == 1
assert head.__dict__["_request_rbo_route_delta"].shape == observed[0].shape
assert torch.allclose(
head.__dict__["_request_rbo_route_delta"],
torch.full_like(observed[0], 0.125),
)
assert torch.equal(
head.__dict__["_request_rbo_route_confidence_t"],
torch.ones(1, 1, 1),
)
def test_request_context_integrates_authenticated_learning_evidence() -> None:
torch.manual_seed(0)
cfg = NexumConfig.tiny()
head = CapacityLMHead(
torch.nn.Linear(cfg.hidden_size, cfg.vocab_size, bias=False),
cfg,
n_layers=1,
)
token_bridge = NexumTBR(cfg)
object.__setattr__(head, "_nexum_token_bridge", token_bridge)
input_ids_t = torch.tensor([[1, 2, 3]], dtype=torch.long)
learned_token_t = token_bridge.token_ids_to_token(
torch.tensor([[1, 2, 3, 4]], dtype=torch.long)
).squeeze(0)
observed_hidden: list[torch.Tensor] = []
observed_rbo_routes: list[torch.Tensor] = []
def record_rbo_route(value: torch.Tensor, _bridge: object) -> torch.Tensor:
observed_rbo_routes.append(value.detach().clone())
return torch.zeros_like(value)
def set_route(
_content: str,
*,
content_hidden: torch.Tensor | None = None,
history_hidden: torch.Tensor | None = None,
) -> None:
assert history_hidden is None
assert content_hidden is not None
observed_hidden.append(content_hidden.detach().clone())
with (
mock.patch.object(head, "set_merkle_route", side_effect=set_route),
mock.patch.object(
head,
"expert_delta_rbo",
side_effect=record_rbo_route,
),
mock.patch.object(
head,
"self_correction_packet",
return_value=SimpleNamespace(confidence=torch.ones(1, 1, 1)),
),
):
head.initialize_request_context(input_ids_t)
assert not bool(head.retained_learning_evidence_rbo_conditioned_t())
head.initialize_request_context(
input_ids_t,
learned_evidence_token_t=learned_token_t,
learned_evidence_episode_length_t=torch.tensor([4]),
learned_evidence_context_length_t=torch.tensor([3]),
)
assert len(observed_hidden) == 2
assert observed_hidden[0].shape[1] == 3
assert observed_hidden[1].shape[1] == 3
assert len(observed_rbo_routes) == 2
assert not torch.equal(observed_rbo_routes[0], observed_rbo_routes[1])
assert bool(head.retained_learning_evidence_rbo_conditioned_t())
expected_learned_hidden = token_bridge.token_to_hidden(learned_token_t.unsqueeze(0))
torch.testing.assert_close(
head.__dict__["_request_learning_key_token_t"],
F.normalize(learned_token_t[:3].sum(dim=0, keepdim=True).unsqueeze(0), dim=-1),
rtol=1.0e-5,
atol=1.0e-6,
)
torch.testing.assert_close(
head.__dict__["_request_learning_value_hidden_t"],
expected_learned_hidden[:, 3:],
rtol=1.0e-5,
atol=1.0e-6,
)
def test_authenticated_current_success_causally_conditions_generation() -> None:
torch.manual_seed(0)
cfg = NexumConfig.tiny()
head = CapacityLMHead(
torch.nn.Linear(cfg.hidden_size, cfg.vocab_size, bias=False),
cfg,
n_layers=1,
)
token_bridge = NexumTBR(cfg)
object.__setattr__(head, "_nexum_token_bridge", token_bridge)
input_ids_t = torch.tensor([[11, 12]], dtype=torch.long)
outcome_token_t = token_bridge.token_ids_to_token(
torch.tensor([[21, 22, 23]], dtype=torch.long)
)
hidden_t = token_bridge.token_to_hidden(
token_bridge.token_ids_to_token(input_ids_t[:, -1:])
)
with (
mock.patch.object(
head,
"expert_delta_rbo",
side_effect=lambda value, _bridge: torch.zeros_like(value),
),
mock.patch.object(
head,
"self_correction_packet",
return_value=SimpleNamespace(confidence=torch.zeros(1, 1, 1)),
),
torch.no_grad(),
):
head.initialize_request_context(input_ids_t)
baseline_logits_t = head(hidden_t)
head.initialize_request_context(
input_ids_t,
grounded_outcome_token_t=outcome_token_t,
grounded_outcome_success_t=torch.tensor(True),
)
grounded_logits_t = head(hidden_t)
assert bool(head.grounded_outcome_evidence_rbo_conditioned_t())
assert int(head.grounded_outcome_evidence_token_count_t()) == 3
assert not torch.equal(grounded_logits_t, baseline_logits_t)
assert isinstance(
head.__dict__["_request_learning_sequence_query_sum_t"],
torch.Tensor,
)
def test_retained_evidence_causally_changes_generation_hidden_state() -> None:
cfg = NexumConfig.tiny()
head = CapacityLMHead(
torch.nn.Linear(cfg.hidden_size, cfg.vocab_size, bias=False),
cfg,
n_layers=1,
)
token_bridge = NexumTBR(cfg)
object.__setattr__(head, "_nexum_token_bridge", token_bridge)
input_ids_t = torch.tensor([[11, 12, 31]], dtype=torch.long)
hidden_t = token_bridge.token_to_hidden(
token_bridge.token_ids_to_token(torch.tensor([[31]], dtype=torch.long))
)
successful_evidence_t = token_bridge.token_ids_to_token(
torch.tensor([[31, 41]], dtype=torch.long)
).squeeze(0)
failed_evidence_t = token_bridge.token_ids_to_token(
torch.tensor([[31, 51]], dtype=torch.long)
).squeeze(0)
projected_hidden: list[torch.Tensor] = []
confidence_inputs: list[tuple[torch.Tensor, torch.Tensor]] = []
hook = head.orig.register_forward_pre_hook(
lambda _module, args: projected_hidden.append(args[0].detach().clone())
)
confidence_hook = head.native_decode_confidence.register_forward_pre_hook(
lambda _module, args: confidence_inputs.append(
(args[0].detach().clone(), args[1].detach().clone())
)
)
try:
with (
mock.patch.object(
head,
"expert_delta_rbo",
side_effect=lambda value, _bridge: torch.zeros_like(value),
),
mock.patch.object(
head,
"self_correction_packet",
return_value=SimpleNamespace(confidence=torch.zeros(1, 1, 1)),
),
torch.no_grad(),
):
head.initialize_request_context(input_ids_t)
head(hidden_t)
head.initialize_request_context(
input_ids_t,
learned_evidence_token_t=successful_evidence_t,
learned_evidence_episode_length_t=torch.tensor([2]),
learned_evidence_context_length_t=torch.tensor([1]),
)
successful_logits_t = head(hidden_t)
head.initialize_request_context(
input_ids_t,
learned_anti_evidence_token_t=failed_evidence_t,
learned_anti_evidence_episode_length_t=torch.tensor([2]),
learned_anti_evidence_context_length_t=torch.tensor([1]),
)
failed_logits_t = head(hidden_t)
head.initialize_request_context(
input_ids_t,
learned_evidence_token_t=successful_evidence_t,
learned_evidence_episode_length_t=torch.tensor([2]),
learned_evidence_context_length_t=torch.tensor([1]),
learned_anti_evidence_token_t=failed_evidence_t,
learned_anti_evidence_episode_length_t=torch.tensor([2]),
learned_anti_evidence_context_length_t=torch.tensor([1]),
)
combined_logits_t = head(hidden_t)
finally:
hook.remove()
confidence_hook.remove()
assert bool(head.retained_learning_evidence_rbo_conditioned_t())
assert len(projected_hidden) == 4
assert len(confidence_inputs) == 4
successful_hidden_t = token_bridge.token_to_hidden(
token_bridge.token_ids_to_token(torch.tensor([[41]], dtype=torch.long))
)
failed_hidden_t = token_bridge.token_to_hidden(
token_bridge.token_ids_to_token(torch.tensor([[51]], dtype=torch.long))
)
baseline_success_distance = (
(projected_hidden[0] - successful_hidden_t).float().pow(2).mean()
)
retained_success_distance = (
(projected_hidden[1] - successful_hidden_t).float().pow(2).mean()
)
baseline_failure_distance = (
(projected_hidden[0] - failed_hidden_t).float().pow(2).mean()
)
retained_failure_distance = (
(projected_hidden[2] - failed_hidden_t).float().pow(2).mean()
)
assert retained_success_distance < baseline_success_distance
assert retained_failure_distance > baseline_failure_distance
assert not torch.equal(successful_logits_t, failed_logits_t)
assert not torch.equal(successful_logits_t, combined_logits_t)
for index, (confidence_hidden_t, confidence_logits_t) in enumerate(
confidence_inputs
):
expected_confidence_hidden_t = head.native_decode_confidence_hidden_t(
projected_hidden[index][:, -1:, :],
confidence_logits_t,
token_bridge,
)
assert confidence_hidden_t.shape == expected_confidence_hidden_t.shape
assert torch.isfinite(confidence_hidden_t).all()
assert not torch.equal(confidence_hidden_t, expected_confidence_hidden_t)
assert confidence_logits_t.shape == (1, 1, cfg.vocab_size)
assert not torch.equal(confidence_inputs[0][0], confidence_inputs[1][0])
assert not torch.equal(confidence_inputs[1][0], confidence_inputs[2][0])
torch.testing.assert_close(confidence_inputs[1][1], successful_logits_t)
torch.testing.assert_close(confidence_inputs[2][1], failed_logits_t)
torch.testing.assert_close(confidence_inputs[3][1], combined_logits_t)
def test_grounded_failure_pressure_preserves_confidence_gated_anti_evidence() -> None:
cfg = NexumConfig.tiny()
head = CapacityLMHead(
torch.nn.Linear(cfg.hidden_size, cfg.vocab_size, bias=True),
cfg,
n_layers=1,
)
token_bridge = NexumTBR(cfg)
object.__setattr__(head, "_nexum_token_bridge", token_bridge)
input_ids_t = torch.tensor([[11, 12, 31]], dtype=torch.long)
failed_episode_ids_t = torch.tensor([[11, 12, 31, 51]], dtype=torch.long)
failed_evidence_t = token_bridge.token_ids_to_token(failed_episode_ids_t).squeeze(0)
routed_hidden: list[torch.Tensor] = []
def record_route(
value: torch.Tensor,
_bridge: NexumTBR,
) -> torch.Tensor:
routed_hidden.append(value.detach().clone())
return torch.zeros_like(value)
with (
mock.patch.object(head, "expert_delta_rbo", side_effect=record_route),
mock.patch.object(
head,
"self_correction_packet",
return_value=SimpleNamespace(confidence=torch.zeros(1, 1, 1)),
),
torch.no_grad(),
):
head.initialize_request_context(
input_ids_t,
learned_anti_evidence_token_t=failed_evidence_t,
learned_anti_evidence_episode_length_t=torch.tensor([4]),
learned_anti_evidence_context_length_t=torch.tensor([3]),
)
unpressured_t = (
head.__dict__["_request_learning_anti_correction_pressure_t"]
.detach()
.clone()
)
correction_bank = ensure_self_correction_bank(head)
correction_bank.grounded_trigger_pressure[0] = 1.0
head.initialize_request_context(
input_ids_t,
learned_anti_evidence_token_t=failed_evidence_t,
learned_anti_evidence_episode_length_t=torch.tensor([4]),
learned_anti_evidence_context_length_t=torch.tensor([3]),
)
pressured_t = (
head.__dict__["_request_learning_anti_correction_pressure_t"]
.detach()
.clone()
)
torch.testing.assert_close(unpressured_t, torch.zeros_like(unpressured_t))
torch.testing.assert_close(pressured_t, torch.ones_like(pressured_t))
assert len(routed_hidden) == 2
torch.testing.assert_close(routed_hidden[0], routed_hidden[1])
def test_unpaired_failure_evidence_cannot_veto_native_action_start() -> None:
cfg = NexumConfig.tiny()
projection = torch.nn.Linear(
cfg.hidden_size,
cfg.vocab_size,
bias=True,
)
head = CapacityLMHead(projection, cfg, n_layers=1)
token_bridge = NexumTBR(cfg)
object.__setattr__(head, "_nexum_token_bridge", token_bridge)
with torch.no_grad():
projection.weight.zero_()
projection.bias.zero_()
projection.bias[41] = 5.0
context_ids_t = torch.tensor([[31]], dtype=torch.long)
failed_action_ids_t = torch.tensor([[41, 51]], dtype=torch.long)
failed_evidence_t = token_bridge.token_ids_to_token(
torch.cat((context_ids_t, failed_action_ids_t), dim=1)
).squeeze(0)
hidden_t = token_bridge.token_to_hidden(
token_bridge.token_ids_to_token(context_ids_t)
)
baseline_logits_t = projection(hidden_t)
with (
mock.patch.object(
head,
"expert_delta_rbo",
side_effect=lambda value, _bridge: torch.zeros_like(value),
),
mock.patch.object(
head,
"self_correction_packet",
return_value=SimpleNamespace(confidence=torch.zeros(1, 1, 1)),
),
torch.no_grad(),
):
head.initialize_request_context(
context_ids_t,
learned_anti_evidence_token_t=failed_evidence_t,
learned_anti_evidence_episode_length_t=torch.tensor([3]),
learned_anti_evidence_context_length_t=torch.tensor([1]),
)
corrected_logits_t = head(hidden_t)
torch.testing.assert_close(corrected_logits_t, baseline_logits_t)
assert int(corrected_logits_t.argmax(dim=-1)) == 41
assert bool(head.retained_learning_evidence_rbo_conditioned_t())
def test_retained_episode_relevance_softly_selects_matching_action() -> None:
cfg = NexumConfig.tiny()
head = CapacityLMHead(
torch.nn.Linear(cfg.hidden_size, cfg.vocab_size, bias=False),
cfg,
n_layers=1,
)
token_bridge = NexumTBR(cfg)
object.__setattr__(head, "_nexum_token_bridge", token_bridge)
request_token_t = torch.zeros(1, 1, cfg.token_dim)
request_token_t[..., 0] = 1.0
matching_action_t = torch.zeros(cfg.token_dim)
matching_action_t[1] = 1.0
unrelated_context_t = -request_token_t.reshape(-1)
unrelated_action_t = torch.zeros(cfg.token_dim)
unrelated_action_t[2] = 1.0
retained_episode_t = torch.stack(
(
request_token_t.reshape(-1),
matching_action_t,
unrelated_context_t,
unrelated_action_t,
),
dim=0,
)
with (
mock.patch.object(
head,
"expert_delta_rbo",
side_effect=lambda value, _bridge: torch.zeros_like(value),
),
mock.patch.object(
head,
"self_correction_packet",
return_value=SimpleNamespace(confidence=torch.zeros(1, 1, 1)),
),
):
head.initialize_request_context(
torch.tensor([[1]], dtype=torch.long),
content_token_t=request_token_t,
learned_evidence_token_t=retained_episode_t,
learned_evidence_episode_length_t=torch.tensor([2, 2]),
learned_evidence_context_length_t=torch.tensor([1, 1]),
)
transition_weight_t = head.__dict__["_request_learning_key_weight_t"]
transition_value_hidden_t = head.__dict__["_request_learning_value_hidden_t"]
assert transition_weight_t.shape == (1, 2, 1)
assert transition_weight_t[0, 0, 0] > transition_weight_t[0, 1, 0]
expected_action_hidden_t = token_bridge.token_to_hidden(
retained_episode_t[[1, 3]].unsqueeze(0)
)
torch.testing.assert_close(
transition_value_hidden_t,
expected_action_hidden_t,
rtol=1.0e-5,
atol=1.0e-6,
)
assert bool(head.retained_learning_evidence_rbo_conditioned_t())
def test_retained_action_replay_is_scoped_to_current_instance_support() -> None:
cfg = NexumConfig.tiny()
token_bridge = NexumTBR(cfg)
retained_context_ids_t = torch.tensor([[10, 11, 12, 13, 14]])
retained_action_ids_t = torch.tensor([[40, 41, 42, 43, 70, 71]])
retained_ids_t = torch.cat(
(retained_context_ids_t, retained_action_ids_t),
dim=1,
)
retained_token_t = token_bridge.token_ids_to_token(retained_ids_t)
retained_hidden_t = token_bridge.token_to_hidden(retained_token_t)
current_instance_ids_t = torch.tensor([[10, 11, 12, 13, 40, 41, 42, 43]])
different_instance_ids_t = torch.tensor([[10, 11, 12, 13, 40, 90, 91, 92]])
contract_fingerprint_t = torch.arange(32, dtype=torch.uint8).reshape(1, 32)
route_args = (
retained_token_t,
retained_hidden_t,
torch.tensor([retained_ids_t.shape[1]]),
torch.tensor([retained_context_ids_t.shape[1]]),
contract_fingerprint_t,
contract_fingerprint_t,
torch.tensor([1]),
)
current_route = _retained_episode_route(
token_bridge.token_ids_to_token(current_instance_ids_t),
*route_args,
)
different_route = _retained_episode_route(
token_bridge.token_ids_to_token(different_instance_ids_t),
*route_args,
)
assert torch.all(current_route.relevance_t > 0)
assert torch.all(different_route.relevance_t > 0)
assert torch.all(current_route.action_replay_relevance_t > 0.4)
assert torch.all(
current_route.action_replay_relevance_t
> different_route.action_replay_relevance_t * 10
)
assert torch.all(
current_route.transition_weight_t.sum(dim=1)
> different_route.transition_weight_t.sum(dim=1) * 10
)
def test_retained_action_replay_selects_one_instance_from_shared_contract() -> None:
cfg = NexumConfig.tiny()
token_bridge = NexumTBR(cfg)
first_context_ids_t = torch.tensor([[10, 11, 12, 13, 14]])
first_action_ids_t = torch.tensor([[40, 50, 51, 52, 53, 70]])
second_context_ids_t = torch.tensor([[20, 21, 22, 23, 24]])
second_action_ids_t = torch.tensor([[40, 60, 61, 62, 63, 80]])
retained_ids_t = torch.cat(
(
first_context_ids_t,
first_action_ids_t,
second_context_ids_t,
second_action_ids_t,
),
dim=1,
)
retained_token_t = token_bridge.token_ids_to_token(retained_ids_t)
contract_fingerprint_t = torch.arange(32, dtype=torch.uint8).reshape(1, 32)
route = _retained_episode_route(
token_bridge.token_ids_to_token(
torch.tensor([[20, 21, 22, 23, 40, 60, 61, 62, 63]])
),
retained_token_t,
token_bridge.token_to_hidden(retained_token_t),
torch.tensor([11, 11]),
torch.tensor([5, 5]),
contract_fingerprint_t,
contract_fingerprint_t.repeat(2, 1),
torch.tensor([1, 1]),
)
first_weight_t = route.transition_weight_t[:, :6, :].sum(dim=1)
second_weight_t = route.transition_weight_t[:, 6:, :].sum(dim=1)
assert torch.all(route.action_replay_relevance_t > 0.4)
assert torch.all(second_weight_t > first_weight_t * 10)
def test_retained_action_replay_requires_shared_objective_instance() -> None:
cfg = NexumConfig.tiny()
token_bridge = NexumTBR(cfg)
retained_context_ids_t = torch.tensor(
[[10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20, 21]]
)
retained_action_ids_t = torch.tensor([[40, 41, 42, 43, 44, 45, 70, 71]])
retained_ids_t = torch.cat(
(retained_context_ids_t, retained_action_ids_t),
dim=1,
)
retained_token_t = token_bridge.token_ids_to_token(retained_ids_t)
contract_fingerprint_t = torch.arange(32, dtype=torch.uint8).reshape(1, 32)
route_args = (
retained_token_t,
token_bridge.token_to_hidden(retained_token_t),
torch.tensor([retained_ids_t.shape[1]]),
torch.tensor([retained_context_ids_t.shape[1]]),
contract_fingerprint_t,
contract_fingerprint_t,
torch.tensor([1]),
)
same_objective_ids_t = torch.tensor(
[
[
90,
91,
12,
13,
14,
15,
16,
17,
18,
19,
40,
41,
42,
43,
44,
45,
70,
71,
]
]
)
shared_generic_objective_ids_t = torch.tensor(
[
[
90,
91,
10,
11,
12,
13,
14,
15,
16,
17,
80,
81,
40,
41,
42,
43,
44,
82,
83,
]
]
)
same_route = _retained_episode_route(
token_bridge.token_ids_to_token(same_objective_ids_t),
*route_args,
)
shared_generic_route = _retained_episode_route(
token_bridge.token_ids_to_token(shared_generic_objective_ids_t),
*route_args,
)
assert torch.all(same_route.action_replay_relevance_t > 0.4)
torch.testing.assert_close(
shared_generic_route.action_replay_relevance_t,
torch.zeros_like(shared_generic_route.action_replay_relevance_t),
)
assert torch.all(same_route.transition_weight_t.sum(dim=1) > 0)
torch.testing.assert_close(
shared_generic_route.transition_weight_t,
torch.zeros_like(shared_generic_route.transition_weight_t),
)
def test_retained_sequence_transitions_recover_repeated_actions_uncapped() -> None:
cfg = NexumConfig.tiny()
token_bridge = NexumTBR(cfg)
context_ids_t = torch.tensor([[7, 8, 9]], dtype=torch.long)
action_ids_t = (
torch.arange(cfg.token_dim * 2 + 3, dtype=torch.long).remainder(5) + 20
).unsqueeze(0)
evidence_ids_t = torch.cat((context_ids_t, action_ids_t), dim=1)
evidence_token_t = token_bridge.token_ids_to_token(evidence_ids_t)
route = _retained_episode_route(
token_bridge.token_ids_to_token(context_ids_t),
evidence_token_t,
token_bridge.token_to_hidden(evidence_token_t),
torch.tensor([evidence_ids_t.shape[1]], dtype=torch.long),
torch.tensor([context_ids_t.shape[1]], dtype=torch.long),
)
assert route.transition_key_token_t.shape[1] == action_ids_t.shape[1]
assert route.transition_key_token_t.shape[1] > cfg.token_dim
torch.testing.assert_close(
route.transition_step_t,
torch.arange(action_ids_t.shape[1], dtype=torch.long),
)
repeated_action_t = action_ids_t[0].eq(action_ids_t[0, 0])
repeated_keys_t = route.transition_key_token_t[:, repeated_action_t, :]
assert repeated_keys_t.shape[1] > 1
assert not torch.equal(repeated_keys_t[:, :1, :], repeated_keys_t[:, 1:2, :])
query_sum_t = route.request_intent_token_t
decoded_ids: list[torch.Tensor] = []
for step in range(action_ids_t.shape[1]):
transition = _retained_evidence_route(
F.normalize(query_sum_t, dim=-1, eps=1.0e-6),
route.transition_key_token_t,
route.transition_value_token_t,
route.transition_weight_t,
)
decoded_id_t = token_bridge.token_to_token_ids(transition.context_t)
decoded_ids.append(decoded_id_t)
position_token_t = native_bit_code(
torch.tensor(step),
cfg.token_dim,
).reshape(1, 1, -1)
query_sum_t = query_sum_t + (
token_bridge.token_ids_to_token(decoded_id_t)
* position_token_t
* query_sum_t.new_tensor(cfg.token_dim).sqrt()
)
torch.testing.assert_close(torch.cat(decoded_ids, dim=1), action_ids_t)
def test_retained_route_confidence_requires_an_active_aligned_arm() -> None:
token_dim = 32
query_t = torch.zeros(1, 1, token_dim)
query_t[..., 0] = 1.0
key_t = torch.zeros(1, 2, token_dim)
key_t[..., 0, 0] = 0.1
key_t[..., 0, 1] = (1.0 - 0.1**2) ** 0.5
key_t[..., 1, 2] = 1.0
value_t = torch.randn(1, 2, token_dim)
weakly_aligned = _retained_evidence_route(
query_t,
key_t,
value_t,
torch.ones(1, 2, 1),
)
assert 0.09 < float(weakly_aligned.confidence_t) < 0.11
exact_key_t = key_t.clone()
exact_key_t[..., 0, :] = query_t
exactly_aligned = _retained_evidence_route(
query_t,
exact_key_t,
value_t,
torch.tensor([[[1.0], [0.0]]]),
)
torch.testing.assert_close(
exactly_aligned.confidence_t,
torch.ones_like(exactly_aligned.confidence_t),
)
exhausted = _retained_evidence_route(
query_t,
exact_key_t,
value_t,
torch.zeros(1, 2, 1),
)
torch.testing.assert_close(
exhausted.confidence_t,
torch.zeros_like(exhausted.confidence_t),
)
def test_retained_failure_preserves_shared_action_prefix_and_rejects_branch() -> None:
vocab_size = 32
shared_base_t = torch.zeros(1, 1, vocab_size)
shared_base_t[..., 7] = 8.0
shared_base_t[..., 11] = 2.0
shared_anti_t = shared_base_t * 3.0 + 4.0
route_t = torch.ones(1, 1, 1)
confidence_t = torch.ones(1, 1, 1)
shared_updated_t = _retained_token_logit_update(
shared_base_t,
route_t,
None,
None,
shared_anti_t,
confidence_t,
)
torch.testing.assert_close(shared_updated_t, shared_base_t)
assert int(shared_updated_t.argmax(dim=-1)) == 7
corrected_branch_t = torch.zeros_like(shared_base_t)
corrected_branch_t[..., 13] = 8.0
corrected_branch_t[..., 17] = 1.0
stale_branch_t = torch.zeros_like(shared_base_t)
stale_branch_t[..., 17] = 8.0
stale_branch_t[..., 13] = 1.0
corrected_updated_t = _retained_token_logit_update(
corrected_branch_t,
route_t,
None,
None,
stale_branch_t,
confidence_t,
)
assert int(corrected_updated_t.argmax(dim=-1)) == 13
assert corrected_updated_t[..., 13] > corrected_updated_t[..., 17]
def test_retained_native_token_route_advances_and_repels_failed_action() -> None:
cfg = NexumConfig.tiny()
projection = torch.nn.Linear(cfg.hidden_size, cfg.vocab_size, bias=False)
head = CapacityLMHead(
projection,
cfg,
n_layers=1,
)
token_bridge = NexumTBR(cfg)
object.__setattr__(head, "_nexum_token_bridge", token_bridge)
with torch.no_grad():
projection.weight.zero_()
context_ids_t = torch.tensor([[31]], dtype=torch.long)
successful_action_ids_t = torch.tensor([[41, 42, 41]], dtype=torch.long)
failed_action_ids_t = torch.tensor([[51, 52, 51]], dtype=torch.long)
successful_evidence_t = token_bridge.token_ids_to_token(
torch.cat((context_ids_t, successful_action_ids_t), dim=1)
).squeeze(0)
failed_evidence_t = token_bridge.token_ids_to_token(
torch.cat((context_ids_t, failed_action_ids_t), dim=1)
).squeeze(0)
hidden_t = token_bridge.token_to_hidden(
token_bridge.token_ids_to_token(context_ids_t)
)
with (
mock.patch.object(
head,
"expert_delta_rbo",
side_effect=lambda value, _bridge: torch.zeros_like(value),
),
mock.patch.object(
head,
"self_correction_packet",
return_value=SimpleNamespace(confidence=torch.zeros(1, 1, 1)),
),
torch.no_grad(),
):
head.initialize_request_context(
context_ids_t,
learned_evidence_token_t=successful_evidence_t,
learned_evidence_episode_length_t=torch.tensor([4]),
learned_evidence_context_length_t=torch.tensor([1]),
learned_anti_evidence_token_t=failed_evidence_t,
learned_anti_evidence_episode_length_t=torch.tensor([4]),
learned_anti_evidence_context_length_t=torch.tensor([1]),
)
logits_t = torch.cat([head(hidden_t) for _ in range(4)], dim=1)
base_logits_t = head.orig(hidden_t)
torch.testing.assert_close(
logits_t[:, :3, :].argmax(dim=-1),
successful_action_ids_t,
)
assert torch.all(
logits_t[:, :3, :].gather(-1, successful_action_ids_t.unsqueeze(-1))
> logits_t[:, :3, :].gather(-1, failed_action_ids_t.unsqueeze(-1))
)
torch.testing.assert_close(logits_t[:, 3:, :], base_logits_t)
torch.testing.assert_close(
head.__dict__["_request_learning_transition_step_t"],
torch.arange(3, dtype=torch.long),
)
torch.testing.assert_close(
head.__dict__["_request_learning_anti_transition_step_t"],
torch.arange(3, dtype=torch.long),
)
def test_retained_success_preserves_shared_prefix_then_rejects_failed_branch() -> None:
cfg = NexumConfig.tiny()
projection = torch.nn.Linear(cfg.hidden_size, cfg.vocab_size, bias=False)
head = CapacityLMHead(
projection,
cfg,
n_layers=1,
)
token_bridge = NexumTBR(cfg)
object.__setattr__(head, "_nexum_token_bridge", token_bridge)
with torch.no_grad():
projection.weight.zero_()
context_ids_t = torch.tensor([[11, 12]], dtype=torch.long)
successful_action_ids_t = torch.tensor([[21, 22]], dtype=torch.long)
failed_action_ids_t = torch.tensor([[21, 23]], dtype=torch.long)
successful_evidence_t = token_bridge.token_ids_to_token(
torch.cat((context_ids_t, successful_action_ids_t), dim=1)
).squeeze(0)
failed_evidence_t = token_bridge.token_ids_to_token(
torch.cat((context_ids_t, failed_action_ids_t), dim=1)
).squeeze(0)
hidden_t = token_bridge.token_to_hidden(
token_bridge.token_ids_to_token(context_ids_t[:, -1:])
)
with (
mock.patch.object(
head,
"expert_delta_rbo",
side_effect=lambda value, _bridge: torch.zeros_like(value),
),
mock.patch.object(
head,
"self_correction_packet",
return_value=SimpleNamespace(confidence=torch.zeros(1, 1, 1)),
),
torch.no_grad(),
):
head.initialize_request_context(
context_ids_t,
learned_evidence_token_t=successful_evidence_t,
learned_evidence_episode_length_t=torch.tensor([4]),
learned_evidence_context_length_t=torch.tensor([2]),
learned_anti_evidence_token_t=failed_evidence_t,
learned_anti_evidence_episode_length_t=torch.tensor([4]),
learned_anti_evidence_context_length_t=torch.tensor([2]),
)
logits_t = torch.cat([head(hidden_t) for _ in range(2)], dim=1)
torch.testing.assert_close(
logits_t.argmax(dim=-1),
successful_action_ids_t,
)
assert logits_t[0, 1, 22] > logits_t[0, 1, 23]
assert bool(head.retained_learning_evidence_rbo_conditioned_t())
def test_verified_success_releases_current_action_and_retains_future_transfer() -> None:
cfg = NexumConfig.tiny()
projection = torch.nn.Linear(cfg.hidden_size, cfg.vocab_size, bias=False)
head = CapacityLMHead(
projection,
cfg,
n_layers=1,
)
token_bridge = NexumTBR(cfg)
object.__setattr__(head, "_nexum_token_bridge", token_bridge)
with torch.no_grad():
projection.weight.zero_()
context_ids_t = torch.tensor([[11, 12]], dtype=torch.long)
successful_action_ids_t = torch.tensor([[21, 22]], dtype=torch.long)
successful_evidence_t = token_bridge.token_ids_to_token(
torch.cat((context_ids_t, successful_action_ids_t), dim=1)
).squeeze(0)
hidden_t = token_bridge.token_to_hidden(
token_bridge.token_ids_to_token(context_ids_t[:, -1:])
)
base_logits_t = projection(hidden_t)
with (
mock.patch.object(
head,
"expert_delta_rbo",
side_effect=lambda value, _bridge: torch.zeros_like(value),
),
mock.patch.object(
head,
"self_correction_packet",
return_value=SimpleNamespace(confidence=torch.zeros(1, 1, 1)),
),
torch.no_grad(),
):
head.initialize_request_context(
context_ids_t,
learned_evidence_token_t=successful_evidence_t,
learned_evidence_episode_length_t=torch.tensor([4]),
learned_evidence_context_length_t=torch.tensor([2]),
learned_transition_open_t=torch.tensor(False),
)
resolved_logits_t = torch.cat([head(hidden_t) for _ in range(2)], dim=1)
head.initialize_request_context(
context_ids_t,
learned_evidence_token_t=successful_evidence_t,
learned_evidence_episode_length_t=torch.tensor([4]),
learned_evidence_context_length_t=torch.tensor([2]),
)
future_logits_t = torch.cat([head(hidden_t) for _ in range(2)], dim=1)
torch.testing.assert_close(
resolved_logits_t,
base_logits_t.expand(-1, 2, -1),
)
torch.testing.assert_close(
future_logits_t.argmax(dim=-1),
successful_action_ids_t,
)
assert bool(head.retained_learning_evidence_rbo_conditioned_t())
def test_recursive_route_refreshes_once_per_generated_token_phase() -> None:
cfg = NexumConfig.tiny()
head = CapacityLMHead(
torch.nn.Linear(cfg.hidden_size, cfg.vocab_size, bias=False),
cfg,
n_layers=1,
)
token_bridge = cast(NexumTBR, SimpleNamespace())
refreshes: list[torch.Tensor] = []
recursive_calls: list[torch.Tensor] = []
def refresh(value: torch.Tensor) -> None:
refreshes.append(value.clone())
def recursive_forward(value: torch.Tensor) -> SimpleNamespace:
recursive_calls.append(value.clone())
return SimpleNamespace(shaped_hidden=value + 0.25, proof={})
hidden = torch.randn(1, 1, cfg.hidden_size)
recursive = SimpleNamespace(forward=recursive_forward)
object.__setattr__(head, "_identity_registry", None)
with (
mock.patch.object(
head,
"_apply_self_correction",
side_effect=lambda value: (value, None),
),
mock.patch.object(head, "_refresh_merkle_from_hidden", side_effect=refresh),
mock.patch.object(head, "_ensure_rbo", return_value=recursive),
):
delta = head.expert_delta_rbo(hidden, token_bridge)
assert len(refreshes) == 1
assert len(recursive_calls) == 1
assert torch.equal(refreshes[0], hidden)
assert torch.equal(recursive_calls[0], hidden)
assert torch.allclose(delta, torch.full_like(hidden, 0.25))
def test_request_context_uses_packaged_graph_strength_without_host_bias() -> None:
cfg = NexumConfig.tiny()
head = CapacityLMHead(
torch.nn.Linear(cfg.hidden_size, cfg.vocab_size, bias=False),
cfg,
n_layers=1,
)
object.__setattr__(head, "_nexum_token_bridge", NexumTBR(cfg))
input_ids_t = torch.tensor([[1, 2, 3]], dtype=torch.long)
head.initialize_request_context(input_ids_t)
graph = head.__dict__["_merkle_graph"]
with torch.no_grad():
graph.anchor_gate.fill_(-20.0)
head.initialize_request_context(input_ids_t)
low_a_t, low_b_t = head.path_registry().domain_router_bias(
1,
1,
layer_idx=0,
)
with torch.no_grad():
graph.anchor_gate.fill_(20.0)
head.initialize_request_context(input_ids_t)
high_a_t, high_b_t = head.path_registry().domain_router_bias(
1,
1,
layer_idx=0,
)
assert high_a_t.abs().amax() > low_a_t.abs().amax()
assert high_b_t.abs().amax() > low_b_t.abs().amax()
def test_merkle_relationship_and_layer_masks_remain_model_owned() -> None:
cfg = NexumConfig.tiny()
cfg.axis_a_count = 2
cfg.axis_b_count = 2
cfg.experts_per_axis = 1
cfg.merkle_node_slots = 4
registry = TraversalPathRegistry(cfg, n_layers=2)
registry.set_context_route_strength(torch.tensor(0.8))
with torch.no_grad():
registry.merkle_active.fill_(1.0)
registry.merkle_expert_axis_a.zero_()
registry.merkle_expert_axis_b.zero_()
registry.merkle_active_node.zero_()
registry.merkle_co_domain[0] = 1
registry.merkle_co_subdomain[0] = 1
registry.merkle_co_node[0] = 1
registry.merkle_co_layer_mask[0] = 1.0
registry.merkle_edge_pro[0, 1] = 3.0
no_relationship_a, no_relationship_b = registry.domain_router_bias(
1, 1, layer_idx=0
)
with torch.no_grad():
registry.merkle_relationship_active.fill_(1.0)
relationship_a, relationship_b = registry.domain_router_bias(1, 1, layer_idx=0)
assert relationship_a[0, 0, 1] > no_relationship_a[0, 0, 1]
assert relationship_b[0, 0, 1] > no_relationship_b[0, 0, 1]
with torch.no_grad():
registry.merkle_co_layer_mask[0] = 0.0
masked_a, masked_b = registry.domain_router_bias(1, 1, layer_idx=0)
co_domains, co_subdomains, co_nodes = registry._merkle_co_targets_for_layer(0)
assert torch.equal(masked_a, no_relationship_a)
assert torch.equal(masked_b, no_relationship_b)
assert torch.all(co_domains == -1)
assert torch.all(co_subdomains == -1)
assert torch.all(co_nodes == -1)
with torch.no_grad():
registry.merkle_layer_order.copy_(torch.tensor([1, 0]))
registry.merkle_active.zero_()
domain_weights = registry.traversal_layer_weights()
registry.merkle_active.fill_(1.0)
merkle_weights = registry.traversal_layer_weights()
assert torch.allclose(domain_weights.sum(), torch.tensor(1.0))
assert torch.allclose(merkle_weights, torch.tensor([1.0 / 3.0, 2.0 / 3.0]))
def test_traversal_registry_hot_routes_do_not_extract_tensor_scalars() -> None:
cfg = NexumConfig.tiny()
cfg.axis_a_count = 2
cfg.axis_b_count = 2
cfg.experts_per_axis = 1
cfg.merkle_node_slots = 4
registry = TraversalPathRegistry(cfg, n_layers=2)
registry.set_context_route_strength(torch.tensor(0.8))
with torch.no_grad():
registry.merkle_active.fill_(1.0)
registry.merkle_relationship_active.fill_(1.0)
registry.merkle_active_node.zero_()
registry.merkle_co_domain[0] = 1
registry.merkle_co_subdomain[0] = 1
registry.merkle_co_node[0] = 1
registry.merkle_co_layer_mask[0] = 1.0
scalar_sync = AssertionError("hot routing extracted a tensor scalar")
with (
mock.patch.object(torch.Tensor, "item", side_effect=scalar_sync),
mock.patch.object(torch.Tensor, "__bool__", side_effect=scalar_sync),
mock.patch.object(torch.Tensor, "__float__", side_effect=scalar_sync),
):
co_targets = registry._merkle_co_targets_for_layer(0)
domain_bias = registry.domain_router_bias(1, 1, layer_idx=0)
layer_weights = registry.traversal_layer_weights()
assert all(value.device == registry.merkle_active.device for value in co_targets)
assert all(value.device == registry.merkle_active.device for value in domain_bias)
assert layer_weights.device == registry.merkle_active.device
def test_layer_dispatch_materializes_model_order_once_without_tensor_item() -> None:
cfg = NexumConfig.tiny()
head = CapacityLMHead(
torch.nn.Linear(cfg.hidden_size, cfg.vocab_size, bias=False),
cfg,
n_layers=2,
)
object.__setattr__(head, "_identity_registry", None)
observed: list[tuple[int, int]] = []
def apply_layer(
hidden: torch.Tensor,
layer_idx: int,
_layer: object,
*,
direction_idx: int,
) -> torch.Tensor:
observed.append((layer_idx, direction_idx))
return hidden
hidden = torch.randn(1, 1, cfg.hidden_size)
scalar_sync = AssertionError("layer dispatch extracted a tensor scalar")
with (
mock.patch.object(head, "_apply_expert_layer", side_effect=apply_layer),
mock.patch.object(head, "self_correction_packet", return_value=None),
mock.patch.object(torch.Tensor, "item", side_effect=scalar_sync),
):
output, path = head._run_layer_pass(
hidden,
direction_idx=0,
layer_indices=torch.tensor([1, 0]),
)
assert observed == [(1, 0), (0, 0)]
assert path == []
assert torch.equal(output, hidden)
def test_outcome_history_changes_next_route_without_host_override() -> None:
torch.manual_seed(7)
cfg = NexumConfig.tiny()
cfg.hop_layer_slots = 2
head = CapacityLMHead(
torch.nn.Linear(cfg.hidden_size, cfg.vocab_size, bias=False),
cfg,
n_layers=2,
)
ensure_domain_router(head)
history = ensure_outcome_history(head)
hidden_t = torch.randn(1, 1, cfg.hidden_size)
before_t = head._none_paged_layer_scale(hidden_t)
history.observe(torch.ones(cfg.evolution_feature_dim), score_norm=0.0)
after_t = head._none_paged_layer_scale(hidden_t)
assert not torch.equal(after_t, before_t)
def test_paged_authority_runs_inside_recursive_feedback_loop() -> None:
cfg = NexumConfig.tiny()
head = CapacityLMHead(
torch.nn.Linear(cfg.hidden_size, cfg.vocab_size, bias=False),
cfg,
n_layers=1,
)
hidden = torch.randn(1, 1, cfg.hidden_size)
stack_calls = 0
ordinary_calls = 0
class TokenBridge:
@staticmethod
def hidden_to_token(value: torch.Tensor) -> torch.Tensor:
return value.new_zeros(*value.shape[:-1], cfg.token_dim)
class Stack:
@staticmethod
def source_bank_indices(device: torch.device | str) -> torch.Tensor:
return torch.zeros(1, device=device, dtype=torch.long)
def __call__(
self,
flat_hidden: torch.Tensor,
_bit_language: torch.Tensor,
_expert_bias: torch.Tensor,
layer_scale: torch.Tensor,
_capacity_rank: int,
*,
correction_pressure_t: torch.Tensor | None = None,
release_confidence_t: torch.Tensor | None = None,
) -> SimpleNamespace:
nonlocal stack_calls
stack_calls += 1
assert layer_scale.shape == (1,)
assert isinstance(correction_pressure_t, torch.Tensor)
assert isinstance(release_confidence_t, torch.Tensor)
return SimpleNamespace(
output_t=flat_hidden + 0.1,
actual_route_pair_ids_t=torch.tensor([[[0, 1]]]),
actual_route_weight_t=torch.tensor([[[0.25, 0.75]]]),
draft_acceptance_t=torch.ones(1, 1, dtype=torch.bool),
wasted_prefetch_t=torch.zeros(1, dtype=torch.bool),
)
class RecursiveAuthority:
def forward(
self,
value: torch.Tensor,
layer_stack_fn: object = None,
) -> NexumRBOResult:
assert callable(layer_stack_fn)
current = value + layer_stack_fn(value)
return NexumRBOResult(
shaped_hidden=current,
reference_hidden=value,
reference_token=value.new_zeros(1, 1, cfg.token_dim),
steps=1,
stop_reason=1,
proof={
"nexumRBOActive": True,
"rboSteps": 1,
"rboRouteGraphOwned": True,
"rboForwardPasses": 1,
"rboFeedbackPasses": 1,
"rboMutationPasses": 1,
"rboReinforcementPasses": 1,
"rboRotationApplied": False,
"rboRunsInsideModel": True,
},
)
@staticmethod
def record_residency_route_sequence(*_args: object) -> None:
return
recursive = RecursiveAuthority()
token_bridge = cast(NexumTBR, TokenBridge())
def ordinary_route(
value: torch.Tensor,
) -> tuple[torch.Tensor, list[tuple[int, torch.Tensor, torch.Tensor]]]:
nonlocal ordinary_calls
ordinary_calls += 1
return value + 0.05, [(0, torch.tensor([[0, 1]]), torch.tensor([[0.6, 0.4]]))]
with (
mock.patch.object(head, "none_paged_stack", return_value=Stack()),
mock.patch.object(head, "_ordinary_routed_hidden", side_effect=ordinary_route),
mock.patch.object(
head,
"_none_paged_expert_bias",
return_value=torch.zeros(1, 1, 1, 8),
),
mock.patch.object(head, "_ensure_rbo", return_value=recursive),
):
delta = head.expert_delta_rbo(hidden, token_bridge)
proof = head.rbo_proof()
assert stack_calls == 1
assert ordinary_calls == 1
assert delta.shape == hidden.shape
assert proof is not None
assert proof["rboSteps"] == 1
assert proof["nonePagedExpertsActive"] is True
assert proof["routeTraceRecorded"] is True
route_packet = head.rbo_route_packet()
assert route_packet is not None
assert torch.equal(
route_packet.output_delta_norm[-2:] / route_packet.output_delta_norm[-2:].sum(),
torch.tensor([0.25, 0.75]),
)
def test_single_route_applies_base_once_and_page_delta_once() -> None:
hidden_t = torch.tensor([[0.25, -0.5]])
bit_t = torch.zeros(1, 2)
request = NoNEPageRequestPacket(
session_id_t=torch.tensor([1, 2, 3, 4]),
generation_t=torch.tensor(1),
route_pair_ids_t=torch.tensor([7]),
unique_route_pair_ids_t=torch.tensor([7]),
route_position_t=torch.tensor([0]),
priority_t=torch.ones(1),
)
frozen = NoNEFrozenPageBatch(
fc1_weight_t=torch.tensor([[[1.0, 0.0], [0.0, 1.0], [0.5, 0.0], [0.0, 0.5]]]),
fc2_weight_t=torch.tensor([[[1.0, 0.0], [0.0, 1.0]]]),
route_pair_ids_t=torch.tensor([7]),
catalog_revision_t=torch.tensor([1]),
)
active = cast(
NoNEActivePageParameters,
SimpleNamespace(
route_pair_ids_t=torch.tensor([7]),
expert_to_token_weight_t=torch.zeros(1, 2, 2),
token_to_expert_weight_t=torch.zeros(1, 2, 2),
bit_to_token_weight_t=torch.zeros(1, 2, 2),
token_to_bit_weight_t=torch.zeros(1, 2, 2),
route_embedding_t=torch.zeros(1, 2),
outcome_memory_t=torch.zeros(1, 2),
knowledge_transfer_t=torch.zeros(1, 2),
verification_embedding_t=torch.zeros(1, 2),
repair_embedding_t=torch.zeros(1, 2),
translation_gate_t=torch.full((1, 1), -100.0),
residual_scale_t=torch.full((1, 1), -100.0),
bit_reconstruction_gate_t=torch.full((1, 1), -100.0),
fc1_down_t=torch.zeros(1, 1, 2),
fc1_up_t=torch.zeros(1, 4, 1),
fc2_down_t=torch.zeros(1, 1, 2),
fc2_up_t=torch.zeros(1, 2, 1),
),
)
packet = NoNEPagedExpertCompute()(
hidden_t,
bit_t,
request,
frozen,
active,
)
assert torch.count_nonzero(packet.output_t) == 0
assert torch.all(packet.route_execution_t)
invalid_request = NoNEPageRequestPacket(
session_id_t=request.session_id_t,
generation_t=request.generation_t,
route_pair_ids_t=request.route_pair_ids_t,
unique_route_pair_ids_t=request.unique_route_pair_ids_t,
route_position_t=torch.tensor([1]),
priority_t=request.priority_t,
)
with pytest.raises(AssertionError, match="did not execute its resident row"):
NoNEPagedExpertCompute()(
hidden_t,
bit_t,
invalid_request,
frozen,
active,
)
def test_paged_layer_restores_execution_receipt_route_shape() -> None:
token_count = 2
top_k = 3
hidden_dim = 4
bit_dim = 2
row_count = token_count * top_k
route_position_t = torch.tensor([0, 1, 2, 0, 1, 2])
route_pair_ids_t = torch.tensor([7, 8, 9, 7, 8, 9])
inner_active = SimpleNamespace(
route_embedding_t=torch.ones(top_k, hidden_dim),
outcome_memory_t=torch.ones(top_k, hidden_dim),
knowledge_transfer_t=torch.ones(top_k, hidden_dim),
verification_embedding_t=torch.ones(top_k, hidden_dim),
repair_embedding_t=torch.zeros(top_k, hidden_dim),
)
class Inner:
def __init__(self) -> None:
self.request = SimpleNamespace(route_position_t=route_position_t)
self.active = inner_active
@staticmethod
def forward(
_hidden_t: torch.Tensor,
_bit_language_t: torch.Tensor,
) -> SimpleNamespace:
return SimpleNamespace(
output_t=torch.zeros(row_count, hidden_dim),
route_execution_t=torch.ones(row_count, dtype=torch.bool),
route_pair_ids_t=route_pair_ids_t,
unique_route_pair_ids_t=torch.tensor([7, 8, 9]),
token_t=torch.zeros(row_count, bit_dim),
reconstructed_bit_language_t=torch.zeros(row_count, bit_dim),
)
transaction = NexumNoNEPagedLayerTransaction(
cast(Any, Inner()),
route_weight_t=torch.full((token_count, top_k), 1.0 / top_k),
hidden_t=torch.zeros(row_count, hidden_dim),
bit_language_t=torch.zeros(row_count, bit_dim),
)
packet = transaction.forward()
assert packet.route_execution_t.shape == (token_count, top_k)
assert torch.all(packet.route_execution_t)
def test_recursive_policy_has_no_host_step_or_traversal_count() -> None:
fields = NexumRBOConfig.__dataclass_fields__
assert "min_steps" not in fields
assert "max_steps" not in fields
assert "plateau_patience" not in fields
def test_recursive_checkpoint_receipt_requires_every_non_residency_key() -> None:
cfg = NexumConfig.tiny()
class Owner(torch.nn.Module):
def __init__(self) -> None:
super().__init__()
self.lm_head = CapacityLMHead(
torch.nn.Linear(cfg.hidden_size, cfg.vocab_size, bias=False),
cfg,
n_layers=1,
)
owner = Owner()
rbo = owner.lm_head._ensure_rbo(cast(NexumTBR, SimpleNamespace()))
state = {
f"rbo.{key}": value.detach().clone()
for key, value in rbo.state_dict().items()
if not key.startswith("residency.")
}
bind_full_state(owner, state)
complete = owner.lm_head.rbo_checkpoint_receipt()
assert complete["state_loaded"] is True
assert complete["keys_loaded"] == complete["keys_expected"]
assert complete["keys_missing"] == 0
state.pop(next(iter(state)))
bind_full_state(owner, state)
incomplete = owner.lm_head.rbo_checkpoint_receipt()
assert incomplete["state_loaded"] is False
assert incomplete["keys_missing"] == 1
def test_route_exhaustion_closes_the_model_owned_frontier_without_a_host_cap(
monkeypatch: pytest.MonkeyPatch,
) -> None:
cfg = NexumConfig.tiny()
head = CapacityLMHead(
torch.nn.Linear(cfg.hidden_size, cfg.vocab_size, bias=False),
cfg,
n_layers=1,
)
token_bridge = NexumTBR(cfg)
rbo = head._ensure_rbo(token_bridge)
route = [
(
0,
torch.zeros(1, 1, 1, dtype=torch.long),
torch.ones(1, 1, 1),
)
]
def never_stop(
positive_signal: torch.Tensor,
_negative_signal: torch.Tensor,
**_signals: torch.Tensor,
) -> tuple[torch.Tensor, torch.Tensor]:
return (
torch.zeros((), device=positive_signal.device, dtype=torch.bool),
torch.zeros((), device=positive_signal.device, dtype=torch.long),
)
monkeypatch.setattr(rbo, "_stop_gate", never_stop)
monkeypatch.setattr(rbo, "_collect_traversal_path", lambda: route)
hidden = torch.randn(1, 1, cfg.hidden_size)
result = rbo.forward(hidden, layer_stack_fn=lambda current: current * 0.01)
assert result.steps == 2
assert result.stop_reason == STOP_REASON_ROUTE_ARMS_EXHAUSTED
assert result.proof["rboArtificialHopCapActive"] is False
assert result.proof["rboActualStopOwner"] == "model_owned_route_frontier_exhaustion"
assert result.proof["rboVisitedRouteArmCount"] == 1
assert result.proof["rboRouteFrontierExpansionCount"] == 1
def test_generation_engagement_requires_real_recursion_and_traversal() -> None:
route_ids = torch.tensor([[[0, 1, 2]], [[3, 4, 5]], [[6, 7, 8]]])
traversal = [
(
0,
route_ids[0],
torch.full_like(route_ids[0], 1.0 / 3.0, dtype=torch.float32),
),
(
1,
route_ids[1],
torch.full_like(route_ids[1], 1.0 / 3.0, dtype=torch.float32),
),
(
2,
route_ids[2],
torch.full_like(route_ids[2], 1.0 / 3.0, dtype=torch.float32),
),
]
class Head:
def __init__(self) -> None:
self._last_none_paged_packet = SimpleNamespace(
actual_route_pair_ids_t=route_ids,
routed_authority_execution_t=torch.ones_like(
route_ids,
dtype=torch.bool,
),
internal_agent_workspace_t=torch.ones(3, 1, 3, 3),
internal_agent_route_mass_t=torch.full((3, 1, 3), 1.0 / 3.0),
internal_agent_active_t=torch.ones(3, 1, 3, dtype=torch.bool),
internal_agent_merged_t=torch.ones(3, 1, 3),
internal_agent_knowledge_t=torch.ones(3, 1, 3, 4),
internal_agent_capability_t=torch.ones(3, 1, 3, 4),
internal_agent_challenge_t=torch.full((3, 1, 3), 0.25),
internal_agent_acceptance_t=torch.full((3, 1, 3), 1.0 / 3.0),
internal_agent_baseline_merged_t=torch.ones(3, 1, 3),
internal_agent_rehearsal_delta_t=torch.zeros(3, 1, 3),
internal_agent_release_confidence_t=torch.full((3, 1), 0.75),
internal_agent_disagreement_t=torch.full((3, 1), 0.25),
internal_agent_correction_pressure_t=torch.zeros(3, 1),
work_role_output_t=torch.ones(3, 1, 5, 3),
work_role_weight_t=torch.full((3, 1, 5), 0.2),
work_role_transfer_t=torch.full((3, 1, 5, 5), 0.2),
drafting_output_t=torch.ones(3, 1, 3),
work_role_coordination_delta_t=torch.zeros(3, 1, 3),
)
self._last_traversal_path = traversal
self._rbo_generation_forward_count_t = torch.tensor(2)
@staticmethod
def rbo_proof() -> dict[str, object]:
return {
"nexumRBOActive": True,
"nonePagedExpertsActive": True,
"rboSteps": 1,
"rboRouteGraphOwned": True,
"routeTraceRecorded": True,
"rboForwardPasses": 1,
"rboFeedbackPasses": 1,
"rboMutationPasses": 1,
"rboReinforcementPasses": 1,
"rboRotationApplied": False,
"rboRunsInsideModel": True,
}
@staticmethod
def self_correction_packet() -> object:
return object()
receipt = _full_generation_engagement(
SimpleNamespace(lm_head=Head()),
expected_generation_forwards=2,
)
assert receipt["recursive_steps"] == 1
assert receipt["recursive_feedback_passes"] == 1
assert receipt["recursive_generation_forwards"] == 2
assert receipt["traversal_layers"] == 3
assert receipt["routed_authority_count"] == 9
assert receipt["routed_authority_execution_count"] == 9
assert receipt["internal_multi_agent_engaged"] is True
assert receipt["internal_agent_banks"] == 3
assert receipt["internal_agent_arms"] == 3
assert receipt["internal_agent_active_arms"] == 3
assert receipt["internal_agent_workspace_contribution_count"] == 9
assert receipt["internal_expert_routes"] == 9
assert receipt["internal_agent_growth_topology_owned"] is True
assert receipt["internal_agent_knowledge_transfer_engaged"] is True
assert receipt["internal_agent_drafting_engaged"] is True
assert receipt["internal_agent_rehearsal_engaged"] is True
assert receipt["internal_agent_challenge_mean"] == pytest.approx(0.25)
assert receipt["internal_agent_release_confidence_mean"] == pytest.approx(0.75)
assert receipt["model_internal_agents_engaged"] is True
assert receipt["model_internal_agent_count"] == 3
assert receipt["model_internal_agent_contribution_count"] == 3
assert receipt["model_internal_authority_count"] == 9
assert receipt["model_internal_authority_execution_count"] == 9
assert receipt["model_internal_bank_count"] == 3
assert receipt["model_internal_worker_count"] == 9
assert receipt["model_internal_worker_execution_count"] == 9
assert receipt["model_internal_worker_contribution_count"] == 9
assert receipt["model_internal_work_roles_engaged"] is True
assert receipt["model_internal_work_role_coordination_applied"] is True
assert receipt["model_internal_work_role_transfer_applied"] is True
assert receipt["model_internal_drafting_engaged"] is True
assert receipt["model_pre_submission_draft_experimented"] is True
assert receipt["model_pre_submission_draft_experiment_count"] == 1
assert (
receipt["model_pre_submission_draft_selection_owner"]
== "trained_rbo_confidence"
)
assert receipt["model_pre_submission_draft_attempt_cap_active"] is False
assert receipt["model_internal_work_role_count"] == 5
assert receipt["model_internal_work_role_contribution_count"] == 5
assert receipt["model_internal_agent_cap_active"] is False
assert receipt["model_internal_worker_cap_active"] is False
assert receipt["model_internal_bank_cap_active"] is False
assert receipt["model_internal_expert_cap_active"] is False
assert receipt["model_internal_host_fanout"] is False
assert receipt["model_rotation_applied"] is False
neutral_packet = Head()._last_none_paged_packet
neutral_packet.internal_agent_workspace_t.zero_()
neutral_packet.internal_agent_merged_t.zero_()
neutral_packet.internal_agent_baseline_merged_t.zero_()
neutral_packet.internal_agent_rehearsal_delta_t.zero_()
neutral_packet.work_role_output_t.zero_()
neutral_packet.drafting_output_t.zero_()
neutral_packet.work_role_coordination_delta_t.zero_()
neutral_head = Head()
neutral_head._last_none_paged_packet = neutral_packet
neutral_receipt = _full_generation_engagement(
SimpleNamespace(lm_head=neutral_head),
expected_generation_forwards=2,
)
assert neutral_receipt["routed_authority_execution_count"] == 9
assert neutral_receipt["internal_agent_workspace_contribution_count"] == 0
assert neutral_receipt["model_internal_worker_contribution_count"] == 0
with mock.patch.object(Head, "rbo_proof", return_value={"nexumRBOActive": True}):
with pytest.raises(RuntimeError, match="model-owned phases"):
_full_generation_engagement(SimpleNamespace(lm_head=Head()))
with pytest.raises(RuntimeError, match="did not match produced tokens"):
_full_generation_engagement(
SimpleNamespace(lm_head=Head()),
expected_generation_forwards=3,
)
@pytest.mark.parametrize("agent_axis_count", (2, 3, 5))
def test_internal_agents_parallelize_trained_routes_without_changing_output(
agent_axis_count: int,
) -> None:
cfg = NexumConfig.tiny()
cfg.hidden_size = 8
cfg.none_intermediate_size = 6
cfg.axis_a_count = 2
cfg.axis_b_count = agent_axis_count
cfg.experts_per_axis = 2
bank = FactorizedGranularExpertBank(cfg)
hidden_t = torch.randn(2, 4, cfg.hidden_size, requires_grad=True)
route_count = cfg.axis_a_count * cfg.axis_b_count * cfg.experts_per_axis
route_ids_t = torch.arange(route_count).reshape(1, 1, -1).expand(2, 4, -1)
route_weight_t = torch.randn(2, 4, route_count).softmax(dim=-1)
expected_t = bank.apply_topk(hidden_t, route_ids_t, route_weight_t)
packet = run_internal_agent_workspaces(
hidden_t,
route_ids_t,
route_weight_t,
bank,
agent_axis_count=agent_axis_count,
experts_per_axis=cfg.experts_per_axis,
)
assert packet.workspace_t.shape == (2, 4, agent_axis_count, cfg.hidden_size)
assert packet.route_mass_t.shape == (2, 4, agent_axis_count)
assert torch.all(packet.active_t)
torch.testing.assert_close(packet.route_mass_t.sum(dim=-1), torch.ones(2, 4))
torch.testing.assert_close(packet.acceptance_t.sum(dim=-1), torch.ones(2, 4))
torch.testing.assert_close(packet.merged_t, expected_t, rtol=1e-5, atol=1e-6)
cast(Any, packet.merged_t.square().mean()).backward()
assert bank.gate_up.grad is not None
assert torch.count_nonzero(bank.gate_up.grad) > 0
@pytest.mark.parametrize(
"bank_type",
(FactorizedGranularExpertBank, MHCExpertBank),
)
def test_selected_authorities_execute_once_with_equivalent_output(
bank_type: type[FactorizedGranularExpertBank] | type[MHCExpertBank],
) -> None:
cfg = NexumConfig.tiny()
cfg.hidden_size = 8
cfg.none_intermediate_size = 6
cfg.axis_a_count = 1
cfg.axis_b_count = 3
cfg.experts_per_axis = 2
bank = bank_type(cfg)
hidden_t = torch.randn(2, 3, cfg.hidden_size)
route_ids_t = torch.tensor(
[
[[0, 3, 5], [2, 2, 1], [4, 0, 3]],
[[5, 1, 0], [3, 4, 2], [1, 5, 5]],
]
)
route_weight_t = torch.randn(2, 3, 3).softmax(dim=-1)
selected_t = bank.apply_selected(hidden_t, route_ids_t)
reference_rows: list[torch.Tensor] = []
for token_hidden_t, token_route_t in zip(
hidden_t.reshape(-1, cfg.hidden_size),
route_ids_t.reshape(-1, route_ids_t.shape[-1]),
strict=True,
):
token_rows: list[torch.Tensor] = []
for route_t in token_route_t:
route_idx = int(route_t)
gate_up_t = F.linear(token_hidden_t, bank.gate_up[route_idx])
gate_t, up_t = gate_up_t.chunk(2, dim=-1)
delta_t = F.linear(F.silu(gate_t) * up_t, bank.down[route_idx])
if isinstance(bank, MHCExpertBank):
delta_t = torch.tanh(delta_t) * torch.sigmoid(bank.alpha[route_idx])
token_rows.append(delta_t)
reference_rows.append(torch.stack(token_rows))
reference_selected_t = torch.stack(reference_rows).reshape_as(selected_t)
reference_merged_t = (reference_selected_t * route_weight_t.unsqueeze(-1)).sum(
dim=-2
)
torch.testing.assert_close(selected_t, reference_selected_t)
torch.testing.assert_close(
bank.apply_topk(hidden_t, route_ids_t, route_weight_t),
reference_merged_t,
)
source = inspect.getsource(bank.apply_selected)
if isinstance(bank, MHCExpertBank):
source += inspect.getsource(bank._apply_selected_chunk)
assert "gate_up.index_select" not in source
assert "down.index_select" not in source
def test_paged_outputs_preserve_independent_internal_agent_workspaces() -> None:
selected_t = torch.randn(3, 6, 7)
route_ids_t = torch.arange(17, 23).reshape(1, 6).expand(3, -1)
route_weight_t = torch.randn(3, 6).softmax(dim=-1)
expected_t = (selected_t * route_weight_t.unsqueeze(-1)).sum(dim=1)
packet = merge_selected_agent_workspaces(
selected_t,
route_ids_t,
route_weight_t,
knowledge_t=selected_t[..., :4],
capability_t=selected_t[..., :5],
)
assert packet.workspace_t.shape == (3, 6, 7)
assert torch.all(packet.active_t)
torch.testing.assert_close(packet.route_mass_t.sum(dim=-1), torch.ones(3))
torch.testing.assert_close(packet.acceptance_t.sum(dim=-1), torch.ones(3))
torch.testing.assert_close(packet.merged_t, expected_t)
assert torch.equal(packet.agent_indices_t, route_ids_t)
def test_internal_agent_rehearsal_uses_correction_pressure_without_excluding_workers() -> (
None
):
selected_t = torch.tensor(
[
[
[1.0, 0.0, 0.0],
[0.9, 0.1, 0.0],
[-1.0, 0.0, 0.0],
[0.0, 1.0, 0.0],
]
]
)
route_ids_t = torch.arange(4).reshape(1, 4)
route_weight_t = torch.tensor([[0.45, 0.35, 0.10, 0.10]])
knowledge_t = torch.tensor([[[0.1], [0.1], [3.0], [0.2]]], dtype=torch.float32)
capability_t = knowledge_t.clone()
first = merge_selected_agent_workspaces(
selected_t,
route_ids_t,
route_weight_t,
knowledge_t=knowledge_t,
capability_t=capability_t,
correction_pressure_t=torch.zeros(()),
release_confidence_t=torch.ones(()),
)
corrected = merge_selected_agent_workspaces(
selected_t,
route_ids_t,
route_weight_t,
knowledge_t=knowledge_t,
capability_t=capability_t,
correction_pressure_t=torch.ones(()),
release_confidence_t=torch.ones(()),
)
assert torch.all(first.acceptance_t > 0)
assert torch.all(corrected.acceptance_t > 0)
torch.testing.assert_close(first.acceptance_t.sum(dim=-1), torch.ones(1))
torch.testing.assert_close(corrected.acceptance_t.sum(dim=-1), torch.ones(1))
assert corrected.acceptance_t[0, 2] > first.acceptance_t[0, 2]
assert not torch.allclose(corrected.merged_t, first.merged_t)
assert torch.count_nonzero(corrected.rehearsal_delta_t) > 0
torch.testing.assert_close(
corrected.correction_pressure_t,
torch.ones_like(corrected.correction_pressure_t),
)
def test_internal_agent_knowledge_transfer_breaks_symmetric_draft_cancellation() -> (
None
):
selected_t = torch.tensor([[[1.0, 0.0], [-1.0, 0.0]]])
route_ids_t = torch.arange(2).reshape(1, 2)
route_weight_t = torch.full((1, 2), 0.5)
directional_evidence_t = torch.tensor([[[1.0, 0.0], [1.0, 0.0]]])
packet = merge_selected_agent_workspaces(
selected_t,
route_ids_t,
route_weight_t,
knowledge_t=directional_evidence_t,
capability_t=directional_evidence_t,
correction_pressure_t=torch.ones(()),
release_confidence_t=torch.ones(()),
)
assert packet.acceptance_t[0, 0] > packet.acceptance_t[0, 1]
assert packet.merged_t[0, 0] > 0
assert torch.count_nonzero(packet.rehearsal_delta_t) > 0
def test_internal_agent_drafting_stays_finite_for_extreme_expert_evidence() -> None:
max_value = torch.finfo(torch.float32).max
selected_t = torch.tensor(
[
[
[max_value, -max_value, float("inf")],
[-max_value, max_value, float("-inf")],
[float("nan"), max_value, -max_value],
[max_value, max_value, -max_value],
]
],
dtype=torch.float32,
)
route_ids_t = torch.arange(4).reshape(1, 4)
route_weight_t = torch.full((1, 4), 0.25)
evidence_t = selected_t[..., :2]
packet = merge_selected_agent_workspaces(
selected_t,
route_ids_t,
route_weight_t,
knowledge_t=evidence_t,
capability_t=evidence_t,
correction_pressure_t=torch.ones(()),
release_confidence_t=torch.ones(()),
)
emitted_tensors = (
packet.workspace_t,
packet.route_mass_t,
packet.merged_t,
packet.knowledge_t,
packet.capability_t,
packet.challenge_t,
packet.acceptance_t,
packet.baseline_merged_t,
packet.rehearsal_delta_t,
packet.release_confidence_t,
packet.disagreement_t,
packet.correction_pressure_t,
packet.work_role_output_t,
packet.work_role_weight_t,
packet.work_role_transfer_t,
packet.drafting_output_t,
packet.work_role_coordination_delta_t,
)
assert all(torch.isfinite(value_t).all() for value_t in emitted_tensors)
torch.testing.assert_close(packet.acceptance_t.sum(dim=-1), torch.ones(1))
torch.testing.assert_close(packet.work_role_weight_t.sum(dim=-1), torch.ones(1))
torch.testing.assert_close(
packet.work_role_transfer_t.sum(dim=-1),
torch.ones(1, 5),
)
def test_authority_head_expands_model_routing_over_every_session_bank() -> None:
cfg = NexumConfig.tiny()
head = CapacityLMHead(
torch.nn.Linear(cfg.hidden_size, cfg.vocab_size, bias=False),
cfg,
n_layers=3,
)
hidden_t = torch.randn(2, 3, cfg.hidden_size)
source_indices_t = torch.tensor([0, 1, 2, 1, 0, 2, 1], dtype=torch.long)
expert_count = cfg.axis_a_count * cfg.axis_b_count * cfg.experts_per_axis
class RoutedStack:
@staticmethod
def source_bank_indices(device: torch.device | str) -> torch.Tensor:
return source_indices_t.to(device)
class RecursiveAuthority:
@staticmethod
def residency_expert_bias(value: torch.Tensor) -> torch.Tensor:
tokens = value.shape[0] * value.shape[1]
return (
torch.arange(3, dtype=torch.float32)
.reshape(3, 1, 1)
.expand(
3,
tokens,
expert_count,
)
)
with mock.patch.object(head, "none_paged_stack", return_value=RoutedStack()):
expanded_bias_t = head._none_paged_expert_bias(
hidden_t,
cast(Any, RecursiveAuthority()),
)
expanded_scale_t = head._none_paged_layer_scale(hidden_t)
assert expanded_bias_t.shape == (
source_indices_t.numel(),
hidden_t.shape[0] * hidden_t.shape[1],
expert_count,
)
assert expanded_scale_t.shape == source_indices_t.shape
torch.testing.assert_close(expanded_bias_t[3], expanded_bias_t[1])
torch.testing.assert_close(expanded_bias_t[4], expanded_bias_t[0])
torch.testing.assert_close(expanded_bias_t[5], expanded_bias_t[2])
torch.testing.assert_close(expanded_scale_t[3], expanded_scale_t[1])
torch.testing.assert_close(expanded_scale_t[4], expanded_scale_t[0])
torch.testing.assert_close(expanded_scale_t[5], expanded_scale_t[2])
def test_route_credit_is_weighted_and_neutral_outcome_is_noop() -> None:
cfg = NexumConfig.tiny()
cfg.axis_a_count = 1
cfg.axis_b_count = 1
cfg.experts_per_axis = 2
registry = TraversalPathRegistry(cfg, n_layers=2)
path = [(0, torch.tensor([0, 1]), torch.tensor([0.25, 0.75]))]
registry.reinforce_traversal_path(
path,
positive_signal=torch.tensor(0.5),
magnitude=torch.tensor(0.0),
)
assert torch.count_nonzero(registry.pro_score) == 0
assert torch.count_nonzero(registry.anti_score) == 0
registry.reinforce_traversal_path(
path,
positive_signal=torch.tensor(1.0),
magnitude=torch.tensor(0.8),
)
assert torch.allclose(registry.pro_score[0, :2], torch.tensor([0.2, 0.6]))
assert torch.count_nonzero(registry.anti_score) == 0
def test_models_endpoint_advertises_only_valid_bundle() -> None:
response: dict[str, object] = {}
class LoadedRunner:
def status(self) -> dict[str, object]:
return {
"loaded": True,
"self_correction_ready": True,
"self_improvement_ready": True,
"context_intent_action_ready": True,
}
class Handler(NexumHandler):
path = "/v1/models"
model_dir = str(MODEL_DIR)
bundle_verified = True
handler = object.__new__(Handler)
with (
mock.patch(
"nexum_runtime.server._json_response",
lambda _handler, status, payload: response.update(
status=status, payload=payload
),
),
mock.patch("nexum_runtime.server._RUNNER", LoadedRunner()),
mock.patch("nexum_runtime.server._RUNNER_LOAD_THREAD", None),
mock.patch("nexum_runtime.server._RUNNER_LOAD_ERROR", ""),
):
handler.do_GET()
assert response["status"] == 200
payload = response["payload"]
assert isinstance(payload, dict)
assert payload["data"][0]["id"] == "Nexum"
def test_health_endpoint_is_minimal_and_uses_startup_proof() -> None:
response: dict[str, object] = {}
class LoadedRunner:
def status(self) -> dict[str, object]:
return {
"loaded": True,
"self_correction_ready": True,
"self_improvement_ready": True,
"context_intent_action_ready": True,
}
class Handler(NexumHandler):
path = "/health"
bundle_verified = True
handler = object.__new__(Handler)
with (
mock.patch(
"nexum_runtime.server._json_response",
lambda _handler, status, payload: response.update(
status=status, payload=payload
),
),
mock.patch("nexum_runtime.server._RUNNER", LoadedRunner()),
):
handler.do_GET()
assert response["status"] == 200
payload = response["payload"]
assert isinstance(payload, dict)
assert payload["ready"] is True
assert "bundle" not in payload
def test_release_status_binds_exact_runtime_and_pristine_namespace() -> None:
response: dict[str, object] = {}
class LoadedRunner:
def status(self) -> dict[str, object]:
return {
"loaded": True,
"self_correction_ready": True,
"self_improvement_ready": True,
"context_intent_action_ready": True,
"context_intent_effect_ready": True,
"context_action_effect_ready": True,
"context_action_conditioning_ready": True,
"full_model_active": True,
"tensor_packages_loaded": 113,
"fast_tokenizer_active": True,
"state_pristine": True,
"learning_generation": 0,
}
class Handler(NexumHandler):
path = "/release-status"
api_key = ""
bundle_verified = True
release_artifact_sha256 = "a" * 64
runtime_source_sha256 = "c" * 64
candidate_identity_sha256 = hashlib.sha256(
(("a" * 64) + ":" + ("c" * 64)).encode("ascii")
).hexdigest()
runtime_instance_sha256 = "b" * 64
state_namespace = "suite:mode:run"
handler = object.__new__(Handler)
with (
mock.patch(
"nexum_runtime.server._json_response",
lambda _handler, status, payload: response.update(
status=status, payload=payload
),
),
mock.patch("nexum_runtime.server._RUNNER", LoadedRunner()),
):
handler.do_GET()
assert response == {
"status": 200,
"payload": {
"schema": "nexum.release-status.v4",
"ready": True,
"full_model_active": True,
"tensor_packages_loaded": 113,
"tokenizer_acceleration_ready": True,
"self_correction_ready": True,
"self_improvement_ready": True,
"context_intent_action_ready": True,
"context_intent_effect_ready": True,
"context_action_effect_ready": True,
"context_action_conditioning_ready": True,
"release_artifact_sha256": "a" * 64,
"runtime_source_sha256": "c" * 64,
"candidate_identity_sha256": hashlib.sha256(
(("a" * 64) + ":" + ("c" * 64)).encode("ascii")
).hexdigest(),
"runtime_instance_sha256": "b" * 64,
"state_namespace": "suite:mode:run",
"state_pristine": True,
"loading": False,
"load_error": "",
"learning_generation": 0,
},
}
def test_self_improvement_status_reports_validated_learning_history() -> None:
response: dict[str, object] = {}
learning = {
"schema": "nexum.learning-status.v1",
"generation": 3,
"decision_chain_sha256": "a" * 64,
"decision_counts": {
"promotion_applied": 2,
"promotion_rejected": 1,
"rollback_applied": 0,
},
"rollback_generations": [0, 1, 2],
"rollback_available": True,
"applies_to": "future_sessions",
}
class LoadedRunner:
def status(self) -> dict[str, object]:
return {
"loaded": True,
"self_improvement_ready": True,
"learning": learning,
}
class Handler(NexumHandler):
path = "/self-improvement/status"
api_key = ""
handler = object.__new__(Handler)
with (
mock.patch(
"nexum_runtime.server._json_response",
lambda _handler, status, payload: response.update(
status=status, payload=payload
),
),
mock.patch("nexum_runtime.server._RUNNER", LoadedRunner()),
):
handler.do_GET()
assert response["status"] == 200
payload = response["payload"]
assert isinstance(payload, dict)
assert payload["active"] is True
assert payload["learning"] == learning
def test_self_improvement_rollback_requires_exact_candidate_and_generation() -> None:
response: dict[str, object] = {}
class Handler(NexumHandler):
path = "/self-improvement/rollback"
model_dir = str(MODEL_DIR)
device = "cpu"
api_key = ""
enable_tools = True
candidate_identity_sha256 = "a" * 64
state_namespace = "release:test"
def _read_payload(self) -> dict[str, object]:
return {
"expected_candidate_identity_sha256": "a" * 64,
"target_generation": 2,
"expected_generation": 3,
}
class StubRunner:
def rollback_learning(
self,
*,
target_generation: int,
expected_generation: int,
) -> dict[str, object]:
assert target_generation == 2
assert expected_generation == 3
return {
"ok": True,
"learning_generation": 4,
"source_generation": 2,
}
handler = object.__new__(Handler)
with (
mock.patch("nexum_runtime.server._runner", return_value=StubRunner()),
mock.patch(
"nexum_runtime.server._json_response",
lambda _handler, status, payload: response.update(
status=status, payload=payload
),
),
):
handler.do_POST()
assert response == {
"status": 200,
"payload": {
"ok": True,
"learning_generation": 4,
"source_generation": 2,
"candidate_identity_sha256": "a" * 64,
"state_namespace": "release:test",
},
}
class StaleHandler(Handler):
def _read_payload(self) -> dict[str, object]:
return {
"expected_candidate_identity_sha256": "b" * 64,
"target_generation": 2,
"expected_generation": 3,
}
response.clear()
stale_handler = object.__new__(StaleHandler)
with mock.patch(
"nexum_runtime.server._json_response",
lambda _handler, status, payload: response.update(
status=status, payload=payload
),
):
stale_handler.do_POST()
assert response == {
"status": 422,
"payload": {
"ok": False,
"error": "ValueError: candidate identity changed before rollback",
},
}
def test_server_does_not_impose_a_fixed_request_payload_cap() -> None:
handler = object.__new__(NexumHandler)
handler.headers = cast(HTTPMessage, {"Content-Length": str(16 * 1024 * 1024 + 1)})
handler.rfile = io.BytesIO(b'{"long_context":true}')
assert handler._read_payload() == {"long_context": True}
def test_chat_completions_endpoint_uses_release_runner() -> None:
response: dict[str, object] = {}
class Handler(NexumHandler):
path = "/v1/chat/completions"
model_dir = str(MODEL_DIR)
device = "cuda:0"
def _read_payload(self) -> dict[str, object]:
return {"model": "Nexum", "messages": [{"role": "user", "content": "hi"}]}
class StubRunner:
def chat(self, payload: dict[str, object]) -> dict[str, object]:
return _completion({"role": "assistant", "content": str(payload["model"])})
handler = object.__new__(Handler)
with (
mock.patch(
"nexum_runtime.server._runner", lambda _model_dir, _device: StubRunner()
),
mock.patch(
"nexum_runtime.server._json_response",
lambda _handler, status, payload: response.update(
status=status, payload=payload
),
),
):
handler.do_POST()
assert response["status"] == 200
payload = response["payload"]
assert isinstance(payload, dict)
assert payload["choices"][0]["message"]["content"] == "Nexum"
def test_responses_conversion_supports_function_calls() -> None:
result = _responses_result(
_completion(
{
"role": "assistant",
"content": None,
"tool_calls": [_tool_call("call_1", "Read", {"path": "README.md"})],
}
)
)
assert result["object"] == "response"
assert result["output"][0]["type"] == "function_call"
assert result["output"][0]["name"] == "Read"
assert result["usage"]["total_tokens"] == 6
def test_responses_conversion_separates_reasoning_from_final_content() -> None:
result = _responses_result(
_completion(
{
"role": "assistant",
"content": "Verified result.",
"reasoning_content": "Inspect privately.",
}
)
)
assert [item["type"] for item in result["output"]] == [
"reasoning",
"message",
]
reasoning = result["output"][0]
assert reasoning["summary"] == [
{"type": "summary_text", "text": "Inspect privately."}
]
assert reasoning["content"] == [
{"type": "reasoning_text", "text": "Inspect privately."}
]
assert result["output"][1]["content"][0]["text"] == "Verified result."
def test_responses_round_trip_preserves_reasoning_for_next_assistant_turn() -> None:
response = _responses_result(
_completion(
{
"role": "assistant",
"content": None,
"reasoning_content": "Inspect the supplied path first.",
"tool_calls": [_tool_call("call_1", "Read", {"path": "README.md"})],
}
)
)
payload = _responses_chat_payload(
{
"model": "Nexum",
"conversation": {"id": "session-1"},
"input": response["output"],
},
)
assert [item["type"] for item in response["output"]] == [
"reasoning",
"function_call",
]
assert payload["messages"] == [
{
"role": "assistant",
"content": "",
"reasoning_content": "Inspect the supplied path first.",
"tool_calls": [
{
"id": "call_1",
"type": "function",
"function": {
"name": "Read",
"arguments": '{"path": "README.md"}',
},
}
],
}
]
def test_responses_input_uses_summary_when_reasoning_text_is_absent() -> None:
payload = _responses_chat_payload(
{
"model": "Nexum",
"input": [
{
"type": "reasoning",
"id": "rs_1",
"summary": [
{
"type": "summary_text",
"text": "Preserve this trajectory.",
}
],
"status": "completed",
},
{
"type": "message",
"role": "assistant",
"content": [{"type": "output_text", "text": "Final content."}],
},
],
}
)
assert payload["messages"][0]["reasoning_content"] == ("Preserve this trajectory.")
assert payload["messages"][0]["content"] == [
{"type": "output_text", "text": "Final content."}
]
def test_responses_input_preserves_function_output_without_trusting_it() -> None:
receipt = {
"name": "Read",
"args": {"path": "README.md"},
"ok": True,
"output": "content",
"executed": True,
}
payload = _responses_chat_payload(
{
"model": "Nexum",
"conversation": {"id": "session-1"},
"input": [
{
"type": "function_call",
"call_id": "call-1",
"name": "Read",
"arguments": '{"path":"README.md"}',
},
{
"type": "function_call_output",
"call_id": "call-1",
"output": json.dumps(receipt),
},
],
}
)
assert payload["session_id"] == "session-1"
assert payload["messages"][1]["role"] == "tool"
assert payload["messages"][1]["tool_call_id"] == "call-1"
assert "nexum_observations" not in payload
observations = _responses_caller_observations(
{
"model": "Nexum",
"conversation": {"id": "session-1"},
"input": [
{
"type": "function_call",
"call_id": "call-1",
"name": "Read",
"arguments": '{"path":"README.md"}',
},
{
"type": "function_call_output",
"call_id": "call-1",
"output": json.dumps(receipt),
},
],
}
)
assert observations[0]["name"] == "Read"
assert observations[0]["args"] == {"path": "README.md"}
assert observations[0]["tool_call_id"] == "call-1"
assert observations[0]["receipt_source"] == "caller_attested"
assert observations[0]["source_trust"] == "caller_owned"
def test_responses_chat_signs_correlated_output_before_native_correction() -> None:
class RecordingRunner:
def __init__(self) -> None:
self.signed: list[tuple[str, dict[str, Any]]] = []
self.payload: dict[str, Any] = {}
def sign_observation(
self, session_id: str, observation: dict[str, Any]
) -> dict[str, Any]:
self.signed.append((session_id, observation))
return {
**observation,
"receipt_nonce": "responses-nonce",
"receipt_auth": "responses-auth",
}
def chat(self, payload: dict[str, Any]) -> dict[str, Any]:
self.payload = payload
return _completion({"role": "assistant", "content": "Recovered."})
runner = RecordingRunner()
result = _responses_chat(
runner,
{
"model": "Nexum",
"conversation": {"id": "responses-session"},
"input": [
{
"type": "function_call",
"call_id": "call-failed",
"name": "Read",
"arguments": '{"path":"missing.txt"}',
},
{
"type": "function_call_output",
"call_id": "call-failed",
"output": json.dumps(
{
"ok": False,
"executed": True,
"output": "",
"error": "missing",
"return_code": 2,
}
),
},
],
},
)
assert result["choices"][0]["message"]["content"] == "Recovered."
assert runner.signed[0][0] == "responses-session"
signed = runner.payload["nexum_observations"][0]
assert signed["ok"] is False
assert signed["exit_code"] == 2
assert signed["error"] == "missing"
assert signed["receipt_nonce"] == "responses-nonce"
def test_non_loopback_server_requires_api_key(tmp_path: Path) -> None:
with pytest.raises(ValueError, match="bearer API key"):
serve(
host="0.0.0.0",
port=0,
model_dir=str(MODEL_DIR),
workspace=str(tmp_path),
device="cpu",
)
def test_tool_enabled_server_requires_api_key_on_loopback(tmp_path: Path) -> None:
with pytest.raises(ValueError, match="server tools"):
serve(
host="127.0.0.1",
port=0,
model_dir=str(MODEL_DIR),
workspace=str(tmp_path),
device="cpu",
enable_tools=True,
)
def test_runtime_health_requires_loaded_model_for_ready() -> None:
status = runtime_health(bundle_ok=True, model_loaded=False)
assert status["ok"] is True
assert status["available"] is True
assert status["ready"] is False
assert status["self_improvement"]["configured_bank_limit"] is None
assert status["self_improvement"]["configured_expert_limit"] is None
assert status["self_improvement"]["new_banks_active_next_forward"] is True
assert status["self_improvement"]["session_topology_resume"] is True
assert runtime_health(bundle_ok=True, model_loaded=True)["ready"] is False
assert (
runtime_health(
bundle_ok=True,
model_loaded=True,
self_correction_ready=True,
self_improvement_ready=True,
context_intent_action_ready=True,
)["ready"]
is True
)
missing_action = runtime_health(
bundle_ok=True,
model_loaded=True,
self_correction_ready=True,
self_improvement_ready=True,
context_intent_action_ready=False,
)
assert missing_action["ready"] is False
assert missing_action["context_intent_action"] == {
"available": True,
"active": False,
"owner": "model",
}
def test_context_intent_action_readiness_probe_requires_both_effects() -> None:
config = NexumArchitectureConfig(
vocab_size=64,
pad_token_id=0,
bos_token_id=1,
eos_token_id=2,
hidden_size=32,
intermediate_size=64,
pathway_intermediate_size=64,
num_hidden_layers=2,
num_attention_heads=4,
num_key_value_heads=2,
num_dense_layers=1,
num_experts=2,
num_experts_per_tok=1,
max_position_embeddings=128,
layer_types=["full_attention", "conv"],
intent_pivot=True,
)
model = torch.nn.Module()
model.attention = NexumAttention(config, layer_idx=0)
ready = _context_intent_action_readiness_probe(model, torch)
assert ready == {
"context_intent_effect_ready": True,
"context_action_effect_ready": True,
"context_action_conditioning_ready": True,
}
assert model.attention.a_proj is not None
with torch.no_grad():
model.attention.a_proj.weight.zero_()
missing_action = _context_intent_action_readiness_probe(model, torch)
assert missing_action["context_intent_effect_ready"] is True
assert missing_action["context_action_effect_ready"] is False
assert missing_action["context_action_conditioning_ready"] is False
def test_runner_status_exposes_only_release_level_load_proof() -> None:
runner = NexumLocalRunner(MODEL_DIR, device="cpu")
runner._loaded = True
runner._model = torch.nn.Linear(1, 1)
runner._none_activation = {
"full_model_active": True,
"tensor_contract": {"tensor_count": 113},
"self_correction_present": True,
"native_decode_confidence_ready": True,
"context_intent_action_ready": True,
"context_intent_effect_ready": True,
"context_action_effect_ready": True,
"context_action_conditioning_ready": True,
"view": "/private/cache/path",
"state_coverage": {"private_surface": 1},
}
status = runner.status()
assert status["full_model_active"] is True
assert status["tensor_packages_loaded"] == 113
assert status["self_correction_ready"] is True
assert status["context_intent_action_ready"] is True
assert status["context_intent_effect_ready"] is True
assert status["context_action_effect_ready"] is True
assert status["context_action_conditioning_ready"] is True
assert "activation" not in status
assert "/private/cache/path" not in json.dumps(status)
def test_release_runner_has_no_generation_head_monkey_patch() -> None:
source = (RELEASE_ROOT / "runtime/src/nexum_runtime/runner.py").read_text(
encoding="utf-8"
)
assert "_install_none_generation_route" not in source
assert 'object.__setattr__(head, "forward"' not in source
def test_generation_uses_one_incremental_native_stop_loop() -> None:
source = (RELEASE_ROOT / "runtime/src/nexum_runtime/runner.py").read_text(
encoding="utf-8"
)
assert '"max_new_tokens": 1' in source
assert "_incremental_uncapped_generate(" in source
assert 'call_kwargs["past_key_values"] = past_key_values' in source
assert 'call_kwargs["return_dict_in_generate"] = True' in source
assert "if client_limit_present:\n if prepared_context" not in source
assert "_NativeSelectedTokenRecorder" in source
def test_incremental_uncapped_generation_preserves_c_a_and_native_stop() -> None:
from transformers import StoppingCriteria, StoppingCriteriaList
class StopAfterTwoTransitions(StoppingCriteria):
def __call__(
self,
input_ids: torch.Tensor,
scores: object,
**kwargs: object,
) -> torch.Tensor:
_ = scores, kwargs
return input_ids.new_full(
(input_ids.shape[0],),
input_ids.shape[-1] >= 6,
dtype=torch.bool,
)
config = NexumArchitectureConfig(
vocab_size=64,
pad_token_id=0,
bos_token_id=1,
eos_token_id=2,
hidden_size=32,
intermediate_size=64,
pathway_intermediate_size=64,
num_hidden_layers=2,
num_attention_heads=4,
num_key_value_heads=2,
num_dense_layers=1,
num_experts=4,
num_experts_per_tok=2,
max_position_embeddings=128,
layer_types=["full_attention", "conv"],
intent_pivot=True,
)
model = NexumForCausalLM(config)
torch.nn.Module.eval(model)
input_ids = torch.tensor([[1, 5, 6, 7]], dtype=torch.long)
criteria = StoppingCriteriaList([StopAfterTwoTransitions()])
model.reset_generation_state()
with torch.inference_mode():
output = _incremental_uncapped_generate(
model,
input_ids=input_ids,
inputs_embeds=None,
generate_kwargs={
"max_new_tokens": 1,
"eos_token_id": None,
"pad_token_id": 0,
"use_cache": True,
"do_sample": False,
"stopping_criteria": criteria,
},
generation_room=32,
physical_tokens=input_ids.shape[-1],
stopping_criteria=criteria,
)
assert output.shape[-1] == 6
attention = model.model.layers[0].self_attn
assert attention._intent_cache is not None
assert attention._action_cache is not None
assert attention._intent_cache.shape[-2] == 5
assert attention._action_cache.shape[-2] == 5
def test_stopping_criteria_resolution_uses_tensor_decision() -> None:
calls: list[tuple[torch.Tensor, object]] = []
def criteria(input_ids: torch.Tensor, scores: object) -> torch.Tensor:
calls.append((input_ids, scores))
return input_ids.new_tensor([True], dtype=torch.bool)
input_ids = torch.tensor([[1, 2, 3]], dtype=torch.long)
assert _stopping_criteria_resolved(criteria, input_ids) is True
assert calls == [(input_ids, None)]
def test_native_path_kernel_dispatch_depends_only_on_tensor_shape() -> None:
config = cast(
NexumArchitectureConfig,
SimpleNamespace(
num_experts=4,
hidden_size=4,
pathway_intermediate_size=3,
),
)
paths = NexumExperts(config)
gated = paths._apply_gate(torch.tensor([[1.0, 2.0, 3.0, 4.0]]))
assert torch.equal(
gated,
torch.nn.functional.silu(gated.new_tensor([[1.0, 2.0]]))
* gated.new_tensor([[3.0, 4.0]]),
)
one_token = torch.ones(1, 4)
four_tokens = torch.ones(4, 4)
one_route = torch.tensor([[0, 1]])
four_routes = one_route.expand(4, -1)
one_weight = torch.full((1, 2), 0.5)
four_weights = one_weight.expand(4, -1)
with torch.no_grad():
paths.gate_up_proj.fill_(0.01)
paths.down_proj.fill_(0.01)
actual = paths(one_token, one_route, one_weight)
assert actual.shape == one_token.shape
assert torch.isfinite(actual).all()
def selected(
_owner: object,
hidden: torch.Tensor,
_indices: torch.Tensor,
_weights: torch.Tensor,
) -> torch.Tensor:
return hidden
with (
mock.patch(
"nexum_runtime.architecture.batched_mm_experts_forward",
side_effect=selected,
) as batched,
mock.patch(
"nexum_runtime.architecture.grouped_mm_experts_forward",
side_effect=selected,
) as grouped,
):
assert torch.equal(paths(one_token, one_route, one_weight), one_token)
batched.assert_called_once()
grouped.assert_not_called()
assert torch.equal(paths(four_tokens, four_routes, four_weights), four_tokens)
grouped.assert_called_once()
def test_grounded_receipt_persists_page_and_future_session_learning() -> None:
events: list[object] = []
route_ids_t = torch.tensor([[[0, 1]]])
route_weights_t = torch.tensor([[[0.4, 0.6]]])
acceptance_t = torch.tensor([[True]])
adaptation_t = torch.tensor(0.75)
class Head(torch.nn.Module):
def __init__(self) -> None:
super().__init__()
self.cfg = NexumConfig.tiny()
class Model(torch.nn.Module):
def __init__(self) -> None:
super().__init__()
self.lm_head = Head()
class SessionBank:
learning_generation = 0
def pending_correction_state(
self, observations: list[dict[str, object]]
) -> tuple[torch.Tensor, list[tuple[int, torch.Tensor, torch.Tensor]]]:
events.append(
(
"correction",
{str(item.get("tool_call_id") or "") for item in observations},
)
)
return torch.ones(3), []
def pending_none_outcome_state(
self, observations: list[dict[str, object]]
) -> object:
events.append(
(
"none",
{str(item.get("tool_call_id") or "") for item in observations},
)
)
return SimpleNamespace(
route_pair_ids_t=route_ids_t,
route_weight_t=route_weights_t,
path_acceptance_t=acceptance_t,
adaptation_t=adaptation_t,
)
def repeated_failure_signal(
self, _observations: list[dict[str, object]]
) -> torch.Tensor:
return torch.tensor(False)
def repeated_success_signal(
self, _observations: list[dict[str, object]]
) -> torch.Tensor:
return torch.tensor(False)
def self_improvement_required_signal(
self, _observations: list[dict[str, object]]
) -> torch.Tensor:
return torch.tensor(True)
def self_improvement_success_signal(
self, _observations: list[dict[str, object]]
) -> torch.Tensor:
return torch.tensor(False)
def capture_grounded_learning_state(self) -> tuple[torch.Tensor, ...]:
events.append("learning_before")
return (torch.zeros(1),)
def commit_observations(self, _observations: list[dict[str, object]]) -> None:
events.append("commit")
def promote_grounded_observation(
self,
record_id: str,
_before: tuple[torch.Tensor, ...],
*,
validation: dict[str, object] | None = None,
evidence_token_t: torch.Tensor | None = None,
evidence_context_length_t: torch.Tensor | None = None,
evidence_contract_fingerprint_t: torch.Tensor | None = None,
evidence_contract_count_t: torch.Tensor | None = None,
) -> dict[str, object]:
assert validation is not None
assert validation["eligible"] is True
assert evidence_token_t is None
assert evidence_context_length_t is None
assert evidence_contract_fingerprint_t is None
assert evidence_contract_count_t is None
events.append(("promote", record_id))
return {
"learning_applied": True,
"learning_generation": 1,
"learning_slots_updated": 2,
"future_session_prior_updated": True,
"learning_state_persisted": True,
"session_isolation_preserved": True,
}
def mark_self_improvement_pending(self) -> None:
events.append("improvement_pending")
receipt = SimpleNamespace(
record_id="a" * 24,
self_improvement_applied=True,
to_dict=lambda: {
"record_id": "a" * 24,
"observations": 1,
"none_paged_updated": True,
"none_paged_generation": 1,
"self_correction_applied": True,
"self_improvement_required": True,
"self_improvement_applied": True,
},
)
raw_observation = {
"name": "Read",
"args": {"path": "missing"},
"ok": False,
"output": "not found",
"executed": True,
"tool_call_id": "call-1",
}
with mock.patch(
"nexum_core.observe_tool_outcomes", return_value=receipt
) as observe:
(
result,
digests,
trajectory_resolved_t,
outcome_evidence,
request_evidence,
) = _observe_grounded_tool_receipts(
Model(),
[raw_observation],
set(),
"session-a",
cast(NexumSessionStateBank, SessionBank()),
)
assert result is not None
assert result["learning_applied"] is True
assert result["learning_generation"] == 1
assert not bool(trajectory_resolved_t)
assert outcome_evidence is None
assert request_evidence is None
assert len(digests) == 1
assert events == [
("correction", {"call-1"}),
("none", {"call-1"}),
"learning_before",
"commit",
("promote", "a" * 24),
"improvement_pending",
]
assert observe.call_args.kwargs["none_route_pair_ids_t"] is route_ids_t
assert observe.call_args.kwargs["none_route_weight_t"] is route_weights_t
assert observe.call_args.kwargs["none_path_acceptance_t"] is acceptance_t
assert observe.call_args.kwargs["none_adaptation_t"] is adaptation_t
assert observe.call_args.kwargs["grounded_evidence_token_t"] is None
assert bool(observe.call_args.kwargs["improvement_required_t"]) is True
def test_grounded_evidence_isolates_untrusted_content_from_persistent_state() -> None:
encoded_texts: list[str] = []
projected_shapes: list[tuple[int, ...]] = []
class Tokenizer:
@staticmethod
def encode(text: str, *, add_special_tokens: bool) -> list[int]:
assert add_special_tokens is False
encoded_texts.append(text)
return [1, 2, 3]
class Bridge:
@staticmethod
def token_ids_to_token(token_ids_t: torch.Tensor) -> torch.Tensor:
return F.one_hot(token_ids_t.remainder(4), num_classes=4).float()
@staticmethod
def token_to_hidden(token_t: torch.Tensor) -> torch.Tensor:
projected_shapes.append(tuple(token_t.shape))
projection_t = torch.arange(24, dtype=torch.float32).reshape(4, 6)
return torch.matmul(token_t, projection_t)
class Head(torch.nn.Module):
def __init__(self) -> None:
super().__init__()
self.anchor = torch.nn.Parameter(torch.zeros(()))
object.__setattr__(self, "_nexum_token_bridge", Bridge())
evidence_t = _grounded_observation_token_evidence(
Head(),
Tokenizer(),
[
{
"name": "Inspect",
"args": {"path": "old"},
"ok": False,
"output": '{"next":{"path":"new"}}',
"executed": True,
"receipt_source": "caller_attested",
"source_trust": "caller_owned",
}
],
)
assert evidence_t is not None
assert evidence_t.retained_token_t is None
assert evidence_t.immediate_hidden_t.shape == (3, 6)
assert projected_shapes == [(1, 3, 4)]
assert len(encoded_texts) == 1
assert '"path":"old"' in encoded_texts[0]
assert '\\"path\\":\\"new\\"' not in encoded_texts[0]
assert '"output_sha256":' in encoded_texts[0]
assert '"output_bytes":23' in encoded_texts[0]
def test_current_action_evidence_excludes_successful_calls_from_anti_evidence() -> None:
successful = {"name": "Read", "ok": True}
failed = {"name": "Write", "ok": False}
sentinel = cast(Any, object())
with mock.patch(
"nexum_runtime.runner._grounded_observation_token_evidence",
return_value=sentinel,
) as build:
result = _current_failure_action_evidence(
object(),
object(),
[successful, failed],
messages=[],
tools=[],
expected_prompt_fingerprint_t=torch.zeros(32, dtype=torch.uint8),
)
assert result is sentinel
assert build.call_args.args[2] == [failed]
assert build.call_args.kwargs["allow_session_local_action_episode"] is True
def test_retained_action_evidence_keeps_success_and_failure_polarity_separate() -> None:
successful = {"name": "Read", "ok": True}
failed = {"name": "Write", "ok": False}
with mock.patch(
"nexum_runtime.runner._grounded_observation_token_evidence",
return_value=None,
) as build:
_trusted_polarized_action_evidence(
object(),
object(),
[successful, failed],
successful=False,
messages=[],
tools=[],
expected_prompt_fingerprint_t=torch.zeros(32, dtype=torch.uint8),
)
failed_call = build.call_args
_trusted_polarized_action_evidence(
object(),
object(),
[successful, failed],
successful=True,
messages=[],
tools=[],
expected_prompt_fingerprint_t=torch.zeros(32, dtype=torch.uint8),
)
successful_call = build.call_args
assert failed_call.args[2] == [failed]
assert successful_call.args[2] == [successful]
assert "allow_session_local_action_episode" not in failed_call.kwargs
assert "allow_session_local_action_episode" not in successful_call.kwargs
def test_grounded_evidence_retains_trusted_runtime_execution_content(
monkeypatch: pytest.MonkeyPatch,
) -> None:
encoded_texts: list[str] = []
secret = "release-secret-value-123"
bearer = "bearer-secret-value-456"
monkeypatch.setenv("NEXUM_TEST_API_TOKEN", secret)
class Tokenizer:
@staticmethod
def encode(text: str, *, add_special_tokens: bool) -> list[int]:
assert add_special_tokens is False
encoded_texts.append(text)
return [1]
class Bridge:
@staticmethod
def token_ids_to_token(token_ids_t: torch.Tensor) -> torch.Tensor:
return F.one_hot(token_ids_t, num_classes=2).float()
@staticmethod
def token_to_hidden(token_t: torch.Tensor) -> torch.Tensor:
return token_t
class Head(torch.nn.Module):
def __init__(self) -> None:
super().__init__()
self.anchor = torch.nn.Parameter(torch.zeros(()))
object.__setattr__(self, "_nexum_token_bridge", Bridge())
evidence_t = _grounded_observation_token_evidence(
Head(),
Tokenizer(),
[
{
"name": "Read",
"args": {
"path": "state.txt",
"api_key": secret,
"endpoint": "https://user:password@example.com/data",
},
"ok": True,
"output": (
f"trusted runtime content {secret} Authorization: Bearer {bearer}"
),
"executed": True,
"receipt_source": "runtime_execution",
"source_trust": "trusted_execution",
}
],
)
assert evidence_t is not None
assert evidence_t.retained_token_t is None
assert "trusted runtime content" in encoded_texts[0]
assert secret not in encoded_texts[0]
assert bearer not in encoded_texts[0]
assert "user:password@" not in encoded_texts[0]
assert encoded_texts[0].count("[REDACTED]") >= 3
validation = _shared_learning_validation(
[
{
"name": "Read",
"args": {"path": "state.txt"},
"ok": True,
"output": "trusted runtime content",
"executed": True,
"receipt_source": "runtime_execution",
"source_trust": "trusted_execution",
}
]
)
assert validation["eligible"] is True
assert validation["trusted_observation_count"] == 1
def test_grounded_failure_keeps_full_feedback_immediate_but_not_durable() -> None:
config = _architecture_config(MODEL_DIR / "config.json")
tokenizer = _load_tokenizer(MODEL_DIR, config)
class Bridge:
@staticmethod
def token_ids_to_token(token_ids_t: torch.Tensor) -> torch.Tensor:
return token_ids_t.to(dtype=torch.float64).unsqueeze(-1)
@staticmethod
def token_to_hidden(token_t: torch.Tensor) -> torch.Tensor:
return token_t.repeat(1, 1, 2)
class Head(torch.nn.Module):
def __init__(self) -> None:
super().__init__()
self.anchor = torch.nn.Parameter(torch.zeros(()))
object.__setattr__(self, "_nexum_token_bridge", Bridge())
tools: list[dict[str, Any]] = [
{
"type": "function",
"function": {
"name": "Bash",
"description": "Execute one workspace command.",
"parameters": {
"type": "object",
"properties": {"command": {"type": "string"}},
"required": ["command"],
"additionalProperties": False,
},
},
},
{
"type": "function",
"function": {
"name": "Distractor",
"description": "Unrelated action that must not enter this episode.",
"parameters": {
"type": "object",
"properties": {"value": {"type": "string"}},
"required": ["value"],
"additionalProperties": False,
},
},
},
]
task_messages: list[dict[str, Any]] = [
{"role": "system", "content": "Use verified workspace evidence."},
{
"role": "user",
"content": "Inspect the active state with the supplied checker.",
},
]
selected_call: dict[str, Any] = {
"id": "call-grounded",
"type": "function",
"function": {
"name": "Bash",
"arguments": json.dumps(
{"command": "stale-command --state old"},
sort_keys=True,
),
},
}
messages: list[dict[str, Any]] = [
*task_messages,
{"role": "assistant", "content": None, "tool_calls": [selected_call]},
{
"role": "tool",
"tool_call_id": "call-grounded",
"name": "Bash",
"content": "authenticated execution failure",
},
]
observation: dict[str, Any] = {
"name": "Bash",
"tool_call_id": "call-grounded",
"args": {"command": "stale-command --state old"},
"ok": False,
"output": (
'{"replacement_arguments":{"command":'
'"current-command --state new"},"reason":"state_changed"}'
),
"executed": True,
"receipt_source": "runtime_execution",
"source_trust": "trusted_execution",
}
prompt_fingerprint_t = _token_id_fingerprint_t(
_prompt_from_messages(tokenizer, task_messages, tools)
)
evidence_t = _grounded_observation_token_evidence(
Head(),
tokenizer,
[observation],
messages=messages,
tools=tools,
expected_prompt_fingerprint_t=prompt_fingerprint_t,
)
assert evidence_t is not None
assert evidence_t.retained_token_t is not None
assert evidence_t.retained_context_length_t is not None
retained_context_tokens = int(evidence_t.retained_context_length_t[0])
assert 0 < retained_context_tokens < evidence_t.retained_token_t.shape[0]
immediate_text = tokenizer.decode(
evidence_t.immediate_hidden_t[:, 0].to(dtype=torch.long).tolist(),
skip_special_tokens=False,
)
retained_text = tokenizer.decode(
evidence_t.retained_token_t[:, 0].to(dtype=torch.long).tolist(),
skip_special_tokens=False,
)
assert "Inspect the active state" in retained_text
assert "Execute one workspace command" not in retained_text
assert 'Bash(command="stale-command --state old")' in retained_text
assert torch.equal(
evidence_t.retained_contract_fingerprint_t,
_tool_contract_fingerprints_t([tools[0]]),
)
assert torch.equal(
evidence_t.retained_contract_count_t,
torch.tensor([1]),
)
assert "Use verified workspace evidence" not in retained_text
assert "Unrelated action" not in retained_text
assert "current-command --state new" not in retained_text
assert "authenticated execution failure" not in retained_text
assert '"output":' in immediate_text
assert "current-command --state new" in immediate_text
caller_observation = {
**observation,
"receipt_source": "caller_attested",
"source_trust": "caller_owned",
}
caller_default_evidence_t = _grounded_observation_token_evidence(
Head(),
tokenizer,
[caller_observation],
messages=messages,
tools=tools,
expected_prompt_fingerprint_t=prompt_fingerprint_t,
)
assert caller_default_evidence_t is not None
assert caller_default_evidence_t.retained_token_t is None
caller_evidence_t = _current_failure_action_evidence(
Head(),
tokenizer,
[caller_observation],
messages=messages,
tools=tools,
expected_prompt_fingerprint_t=prompt_fingerprint_t,
)
assert caller_evidence_t is not None
assert caller_evidence_t.retained_token_t is not None
assert caller_evidence_t.retained_context_length_t is not None
torch.testing.assert_close(
caller_evidence_t.retained_token_t,
evidence_t.retained_token_t,
)
caller_immediate_text = tokenizer.decode(
caller_evidence_t.immediate_hidden_t[:, 0].to(dtype=torch.long).tolist(),
skip_special_tokens=False,
)
assert "current-command --state new" not in caller_immediate_text
assert '"output_sha256":' in caller_immediate_text
caller_validation = _shared_learning_validation([caller_observation])
assert caller_validation["eligible"] is False
assert (
caller_validation["reason"]
== "cross_session_transfer_requires_trusted_runtime_execution"
)
runtime_rejection_evidence_t = _current_failure_action_evidence(
Head(),
tokenizer,
[
{
**observation,
"executed": False,
"receipt_source": "runtime_rejection",
"source_trust": "trusted_execution",
}
],
messages=messages,
tools=tools,
expected_prompt_fingerprint_t=prompt_fingerprint_t,
)
assert runtime_rejection_evidence_t is not None
assert runtime_rejection_evidence_t.retained_token_t is not None
torch.testing.assert_close(
runtime_rejection_evidence_t.retained_token_t,
evidence_t.retained_token_t,
)
unavailable_action_tools = [tools[1]]
unavailable_action_prompt_fingerprint_t = _token_id_fingerprint_t(
_prompt_from_messages(
tokenizer,
task_messages,
unavailable_action_tools,
)
)
unavailable_action_rejection_evidence_t = _current_failure_action_evidence(
Head(),
tokenizer,
[
{
**observation,
"executed": False,
"receipt_source": "runtime_rejection",
"source_trust": "trusted_execution",
}
],
messages=messages,
tools=unavailable_action_tools,
expected_prompt_fingerprint_t=unavailable_action_prompt_fingerprint_t,
)
assert unavailable_action_rejection_evidence_t is not None
assert unavailable_action_rejection_evidence_t.retained_token_t is not None
unavailable_action_retained_text = tokenizer.decode(
unavailable_action_rejection_evidence_t.retained_token_t[:, 0]
.to(dtype=torch.long)
.tolist(),
skip_special_tokens=False,
)
assert 'Bash(command="stale-command --state old")' in (
unavailable_action_retained_text
)
assert "Unrelated action that must not enter this episode" not in (
unavailable_action_retained_text
)
assert "current-command --state new" not in unavailable_action_retained_text
assert "authenticated execution failure" not in unavailable_action_retained_text
with pytest.raises(
ValueError,
match="could not bind every selected tool schema",
):
_grounded_observation_token_evidence(
Head(),
tokenizer,
[observation],
messages=messages,
tools=unavailable_action_tools,
expected_prompt_fingerprint_t=unavailable_action_prompt_fingerprint_t,
)
changed_task_messages: list[dict[str, Any]] = [
task_messages[0],
{"role": "user", "content": "Inspect a different active state."},
]
changed_messages: list[dict[str, Any]] = [
*changed_task_messages,
messages[2],
messages[3],
]
changed_evidence_t = _grounded_observation_token_evidence(
Head(),
tokenizer,
[observation],
messages=changed_messages,
tools=tools,
expected_prompt_fingerprint_t=_token_id_fingerprint_t(
_prompt_from_messages(tokenizer, changed_task_messages, tools)
),
)
assert changed_evidence_t is not None
assert changed_evidence_t.retained_token_t is not None
assert not torch.equal(
evidence_t.retained_token_t,
changed_evidence_t.retained_token_t,
)
with pytest.raises(ValueError, match="does not match the issued model turn"):
_grounded_observation_token_evidence(
Head(),
tokenizer,
[observation],
messages=messages,
tools=tools,
expected_prompt_fingerprint_t=torch.zeros(32, dtype=torch.uint8),
)
def test_retained_action_relevance_is_bound_to_the_supplied_contract() -> None:
config = _architecture_config(MODEL_DIR / "config.json")
tokenizer = _load_tokenizer(MODEL_DIR, config)
token_dim = 168
current_tools = [
{
"type": "function",
"function": {
"name": "CurrentStateAction",
"description": "Apply the exact active state transition.",
"parameters": {
"type": "object",
"properties": {"state_ref": {"type": "string"}},
"required": ["state_ref"],
"additionalProperties": False,
},
},
}
]
prior_tools = [
{
"type": "function",
"function": {
"name": "Bash",
"description": "Execute one workspace command.",
"parameters": {
"type": "object",
"properties": {"command": {"type": "string"}},
"required": ["command"],
"additionalProperties": False,
},
},
}
]
current_messages = [
{"role": "user", "content": "Apply the current active state transition."}
]
prior_messages = [
{"role": "user", "content": "Prepare the exact grounded patch operation."}
]
paraphrased_messages = [
{"role": "user", "content": "Perform the active state transition now."}
]
request_context_t = native_bit_code(
_prompt_from_messages(tokenizer, current_messages, current_tools),
token_dim,
)
current_objective_t = native_bit_code(
_prompt_from_messages(tokenizer, current_messages, None),
token_dim,
)
paraphrased_objective_t = native_bit_code(
_prompt_from_messages(tokenizer, paraphrased_messages, None),
token_dim,
)
prior_objective_t = native_bit_code(
_prompt_from_messages(tokenizer, prior_messages, None),
token_dim,
)
action_token_t = native_bit_code(torch.tensor([[17321]]), token_dim)
current_episode_t = torch.cat((current_objective_t, action_token_t), dim=1)
paraphrased_episode_t = torch.cat(
(paraphrased_objective_t, action_token_t),
dim=1,
)
prior_episode_t = torch.cat((prior_objective_t, action_token_t), dim=1)
current_contract_t = _tool_contract_fingerprints_t(current_tools)
prior_contract_t = _tool_contract_fingerprints_t(prior_tools)
current_route = _retained_episode_route(
request_context_t,
current_episode_t,
current_episode_t,
torch.tensor([current_episode_t.shape[1]]),
torch.tensor([current_objective_t.shape[1]]),
current_contract_t,
current_contract_t,
torch.tensor([1]),
)
paraphrased_route = _retained_episode_route(
request_context_t,
paraphrased_episode_t,
paraphrased_episode_t,
torch.tensor([paraphrased_episode_t.shape[1]]),
torch.tensor([paraphrased_objective_t.shape[1]]),
current_contract_t,
current_contract_t,
torch.tensor([1]),
)
prior_route = _retained_episode_route(
request_context_t,
prior_episode_t,
prior_episode_t,
torch.tensor([prior_episode_t.shape[1]]),
torch.tensor([prior_objective_t.shape[1]]),
current_contract_t,
prior_contract_t,
torch.tensor([1]),
)
assert torch.all(current_route.relevance_t > 0.99)
assert torch.all(paraphrased_route.relevance_t > 0.05)
torch.testing.assert_close(
prior_route.relevance_t,
torch.zeros_like(prior_route.relevance_t),
)
def test_grounded_correction_does_not_promote_future_session_learning() -> None:
events: list[str] = []
class Head(torch.nn.Module):
def __init__(self) -> None:
super().__init__()
self.cfg = NexumConfig.tiny()
class Model(torch.nn.Module):
def __init__(self) -> None:
super().__init__()
self.lm_head = Head()
class SessionBank:
learning_generation = 7
shared_topology_bank_count = 1
def pending_correction_state(
self, _observations: list[dict[str, object]]
) -> tuple[torch.Tensor, list[tuple[int, torch.Tensor, torch.Tensor]]]:
return torch.ones(3), []
def pending_none_outcome_state(
self, _observations: list[dict[str, object]]
) -> None:
return None
def repeated_failure_signal(
self, _observations: list[dict[str, object]]
) -> torch.Tensor:
return torch.tensor(False)
def repeated_success_signal(
self, _observations: list[dict[str, object]]
) -> torch.Tensor:
return torch.tensor(False)
def self_improvement_required_signal(
self, _observations: list[dict[str, object]]
) -> torch.Tensor:
return torch.tensor(False)
def self_improvement_success_signal(
self, _observations: list[dict[str, object]]
) -> torch.Tensor:
return torch.tensor(False)
def capture_grounded_learning_state(self) -> tuple[torch.Tensor, ...]:
return (torch.zeros(1),)
def shared_learning_evidence_token(self) -> torch.Tensor:
return torch.zeros(0, dtype=torch.long)
def shared_learning_anti_evidence_token(self) -> torch.Tensor:
return torch.zeros(0, dtype=torch.long)
def commit_observations(self, _observations: list[dict[str, object]]) -> None:
events.append("commit")
def promote_grounded_observation(
self, _record_id: str, _before: tuple[torch.Tensor, ...]
) -> dict[str, object]:
raise AssertionError("correction-only evidence must not be promoted")
receipt = SimpleNamespace(
record_id="b" * 24,
self_improvement_applied=False,
to_dict=lambda: {
"record_id": "b" * 24,
"observations": 1,
"self_correction_applied": True,
"self_improvement_required": False,
"self_improvement_applied": False,
},
)
observation = {
"name": "Read",
"args": {"path": "missing"},
"ok": False,
"output": "not found",
"executed": True,
"tool_call_id": "call-2",
}
with mock.patch("nexum_core.observe_tool_outcomes", return_value=receipt):
(
result,
digests,
trajectory_resolved_t,
outcome_evidence,
request_evidence,
) = _observe_grounded_tool_receipts(
Model(),
[observation],
set(),
"session-correction",
cast(NexumSessionStateBank, SessionBank()),
)
assert result is not None
assert result["self_correction_applied"] is True
assert result["self_improvement_required"] is False
assert result["self_improvement_applied"] is False
assert result["learning_applied"] is False
assert result["learning_generation"] == 7
assert result["learning_slots_updated"] == 0
assert result["future_session_prior_updated"] is False
assert not bool(trajectory_resolved_t)
assert outcome_evidence is None
assert request_evidence is None
assert events == ["commit"]
assert len(digests) == 1
def test_grounded_success_consolidates_an_open_improvement_trajectory() -> None:
events: list[str] = []
class Head(torch.nn.Module):
def __init__(self) -> None:
super().__init__()
self.cfg = NexumConfig.tiny()
class Model(torch.nn.Module):
def __init__(self) -> None:
super().__init__()
self.lm_head = Head()
class SessionBank:
learning_generation = 7
shared_topology_bank_count = 3
def pending_correction_state(
self, _observations: list[dict[str, object]]
) -> tuple[torch.Tensor, list[tuple[int, torch.Tensor, torch.Tensor]]]:
return torch.ones(3), []
def pending_none_outcome_state(
self, _observations: list[dict[str, object]]
) -> None:
return None
def repeated_failure_signal(
self, _observations: list[dict[str, object]]
) -> torch.Tensor:
return torch.tensor(False)
def repeated_success_signal(
self, _observations: list[dict[str, object]]
) -> torch.Tensor:
return torch.tensor(False)
def self_improvement_required_signal(
self, _observations: list[dict[str, object]]
) -> torch.Tensor:
return torch.tensor(False)
def self_improvement_success_signal(
self, _observations: list[dict[str, object]]
) -> torch.Tensor:
return torch.tensor(True)
def capture_grounded_learning_state(self) -> tuple[torch.Tensor, ...]:
return (torch.zeros(1),)
def shared_learning_evidence_token(self) -> torch.Tensor:
return torch.zeros(4, 6)
def shared_learning_anti_evidence_token(self) -> torch.Tensor:
return torch.zeros(3, 6)
def commit_observations(self, _observations: list[dict[str, object]]) -> None:
events.append("commit")
def consolidate_grounded_success(
self,
record_id: str,
*,
validation: dict[str, object] | None = None,
evidence_token_t: torch.Tensor | None = None,
evidence_context_length_t: torch.Tensor | None = None,
evidence_contract_fingerprint_t: torch.Tensor | None = None,
evidence_contract_count_t: torch.Tensor | None = None,
) -> dict[str, object]:
assert record_id == "e" * 24
assert validation is not None
assert validation["eligible"] is True
assert evidence_token_t is None
assert evidence_context_length_t is None
assert evidence_contract_fingerprint_t is None
assert evidence_contract_count_t is None
events.append("consolidate")
return {
"learning_applied": True,
"learning_generation": 8,
"learning_slots_updated": 0,
"future_session_prior_updated": True,
"learning_state_persisted": True,
"session_isolation_preserved": True,
}
receipt = SimpleNamespace(
record_id="e" * 24,
self_improvement_applied=False,
to_dict=lambda: {
"record_id": "e" * 24,
"observations": 1,
"self_correction_applied": False,
"self_improvement_required": False,
"self_improvement_applied": False,
},
)
observation = {
"name": "Bash",
"args": {"command": "verified-command"},
"ok": True,
"output": "verified",
"executed": True,
"tool_call_id": "call-success",
"receipt_source": "runtime_execution",
"source_trust": "trusted_execution",
}
with mock.patch("nexum_core.observe_tool_outcomes", return_value=receipt):
(
result,
digests,
trajectory_resolved_t,
outcome_evidence,
request_evidence,
) = _observe_grounded_tool_receipts(
Model(),
[observation],
set(),
"session-improved",
cast(NexumSessionStateBank, SessionBank()),
)
assert result is not None
assert result["learning_applied"] is True
assert result["learning_generation"] == 8
assert result["learning_slots_updated"] == 0
assert result["future_session_prior_updated"] is True
assert result["learning_trajectory_resolved"] is True
assert bool(trajectory_resolved_t)
assert outcome_evidence is None
assert request_evidence is None
assert events == ["commit", "consolidate"]
assert len(digests) == 1
def test_nnf_x_trajectory_preserves_grounded_correction_and_improvement() -> None:
first = _accumulate_runtime_trajectory(
{},
{
"full_model_active": True,
"grounded_tool_result_message_bound": True,
"grounded_self_correction_message_bound": True,
"self_correction_applied": True,
"learning_generation": 0,
},
)
improved = _accumulate_runtime_trajectory(
first,
{
"full_model_active": True,
"grounded_tool_result_message_bound": True,
"grounded_correction_failure_detected": True,
"grounded_self_improvement_escalated": True,
"grounded_self_improvement_applied": True,
"grounded_self_improvement_persisted": True,
"grounded_self_improvement_source": ("trained_rbo_none_correction_failure"),
"grounded_self_improvement_scope": "session_and_future_sessions",
"grounded_none_paged_generation": 2710,
"grounded_none_paged_authorities_updated": 84,
"grounded_none_banks_grown": 1,
"grounded_none_bank_count": 8,
"grounded_none_growth_pressure": 1.25,
"learning_generation": 1,
"learning_slots_updated": 4,
"self_improvement_applied": True,
"self_improvement_persisted": True,
"future_session_prior_updated": True,
"learning_state_persisted": True,
"session_state_persisted": True,
},
)
assert improved["grounded_self_correction_message_bound_observed"] is True
assert improved["grounded_self_improvement_applied_observed"] is True
assert improved["grounded_self_improvement_persisted_observed"] is True
assert improved["grounded_none_paged_generation_max"] == 2710
assert improved["grounded_none_paged_authorities_updated_max"] == 84
assert improved["grounded_none_banks_grown_max"] == 1
assert improved["grounded_none_bank_count_max"] == 8
assert improved["grounded_none_growth_pressure_max"] == 1.25
assert improved["learning_generation_max"] == 1
assert improved["learning_slots_updated_max"] == 4
assert (
improved["grounded_self_improvement_source"]
== "trained_rbo_none_correction_failure"
)
assert improved["grounded_self_improvement_scope"] == (
"session_and_future_sessions"
)
def test_executor_receipts_are_session_bound_tamper_evident_and_single_use() -> None:
class SessionBank:
def activate(self, _session_id: str) -> None:
return
def validate_pending_observation(self, observation: dict[str, object]) -> None:
assert observation.get("tool_call_id")
def observation_nonce_consumed(self, _nonce: str) -> bool:
return False
def commit_observation_nonces(self, _nonces: tuple[str, ...]) -> None:
return
runner = NexumLocalRunner(MODEL_DIR, device="cpu")
runner._session_bank = cast(NexumSessionStateBank, SessionBank())
runner.load = lambda: {} # type: ignore[method-assign]
observation = {
"name": "Read",
"args": {"path": "README.md"},
"ok": True,
"output": "content",
"executed": True,
"tool_call_id": "call-1",
"receipt_source": "runtime_execution",
"source_trust": "trusted_execution",
}
signed = runner.sign_observation("session-a", observation)
verified, nonces = runner._verified_observations("session-a", [signed])
assert verified == [observation]
retry_verified, retry_nonces = runner._verified_observations("session-a", [signed])
assert retry_verified == verified
assert retry_nonces == nonces
runner._commit_observation_nonces("session-a", nonces)
consumed_rows, consumed_nonces = runner._verified_observations(
"session-a", [signed]
)
assert consumed_rows == []
assert consumed_nonces == ()
tampered = runner.sign_observation("session-b", observation)
tampered["output"] = "forged"
with pytest.raises(ValueError, match="authentication failed"):
runner._verified_observations("session-b", [tampered])
with pytest.raises(ValueError, match="authentication is required"):
runner._verified_observations("session-c", [observation])
with pytest.raises(ValueError, match="missing required fields"):
runner.sign_observation(
"session-d",
{"name": "Read", "args": {}, "ok": True, "output": "content"},
)
rejected = runner.sign_observation(
"session-e",
{
"name": "Bash",
"args": {},
"ok": False,
"output": "invalid arguments",
"executed": False,
"tool_call_id": "call-rejected",
"receipt_source": "runtime_rejection",
"source_trust": "trusted_execution",
},
)
rejected_rows, rejected_nonces = runner._verified_observations(
"session-e", [rejected]
)
assert rejected_rows is not None and rejected_rows[0]["ok"] is False
runner._commit_observation_nonces("session-e", rejected_nonces)
with pytest.raises(ValueError, match="cannot be successful"):
runner.sign_observation(
"session-f",
{
"name": "Bash",
"args": {},
"ok": True,
"output": "",
"executed": False,
"tool_call_id": "call-invalid",
"receipt_source": "runtime_rejection",
"source_trust": "trusted_execution",
},
)
def test_executor_receipt_auth_survives_runner_restart(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
state_root = tmp_path / "state"
monkeypatch.setenv("NEXUM_STATE_DIR", str(state_root))
observation = {
"name": "Read",
"args": {"path": "README.md"},
"ok": True,
"output": "content",
"executed": True,
"tool_call_id": "call-restart",
"receipt_source": "runtime_execution",
"source_trust": "trusted_execution",
}
before = NexumLocalRunner(MODEL_DIR, device="cpu")
before._session_bank = cast(
NexumSessionStateBank,
SimpleNamespace(
activate=lambda _session_id: None,
validate_pending_observation=lambda _observation: None,
),
)
before.load = lambda: {} # type: ignore[method-assign]
signed = before.sign_observation("session-restart", observation)
after = NexumLocalRunner(MODEL_DIR, device="cpu")
verified, nonces = after._verified_observations("session-restart", [signed])
assert verified == [observation]
after._commit_observation_nonces("session-restart", nonces)
consumed_rows, consumed_nonces = after._verified_observations(
"session-restart", [signed]
)
assert consumed_rows == []
assert consumed_nonces == ()
key_path = state_root / "auth" / "receipt.key"
assert key_path.stat().st_mode & 0o777 == 0o600
def test_server_starts_validated_model_load_before_listening(tmp_path: Path) -> None:
events: list[str] = []
class StubRunner:
pass
class StubServer:
def __init__(self, _address: object, _handler: object) -> None:
events.append("bind")
def serve_forever(self) -> None:
events.append("serve")
with (
mock.patch(
"nexum_runtime.server.validate_bundle",
return_value=SimpleNamespace(ok=True, tensor_hashes_verified=True),
),
mock.patch(
"nexum_runtime.server.release_artifact_sha256",
return_value="a" * 64,
),
mock.patch(
"nexum_runtime.server._runtime_instance_sha256",
return_value="b" * 64,
),
mock.patch(
"nexum_runtime.server._runner",
return_value=StubRunner(),
),
mock.patch(
"nexum_runtime.server._start_runner_load",
side_effect=lambda _runner: events.append("load-start"),
),
mock.patch(
"nexum_runtime.server.ThreadingHTTPServer",
StubServer,
),
):
serve(
host="127.0.0.1",
port=0,
model_dir=str(MODEL_DIR),
workspace=str(tmp_path),
device="cpu",
)
assert events == ["load-start", "bind", "serve"]
def test_server_tool_execution_is_disabled_by_default() -> None:
response: dict[str, object] = {}
class Handler(NexumHandler):
path = "/tools/execute"
api_key = ""
enable_tools = False
def _read_payload(self) -> dict[str, object]:
return {"text": "Bash(command='true')"}
handler = object.__new__(Handler)
with mock.patch(
"nexum_runtime.server._json_response",
lambda _handler, status, payload: response.update(
status=status, payload=payload
),
):
handler.do_POST()
assert response["status"] == 403
assert response["payload"] == {
"ok": False,
"error": "server_tool_execution_disabled",
}
def test_server_signs_results_it_executes_with_runtime_provenance(
tmp_path: Path,
) -> None:
response: dict[str, object] = {}
executed = ToolExecutionResult(
name="Bash",
args={"command": "printf verified"},
ok=True,
tool_call_id="call-1",
stdout="verified",
executed=True,
source_trust="trusted_execution",
)
result_payload = executed.to_dict()
expected_observation = {
**result_payload,
"receipt_source": "runtime_execution",
}
class Handler(NexumHandler):
path = "/tools/execute"
model_dir = str(MODEL_DIR)
device = "cpu"
api_key = ""
enable_tools = True
workspace = str(tmp_path)
def _read_payload(self) -> dict[str, object]:
return {
"tool_calls": [
{
"id": "call-1",
"type": "function",
"function": {
"name": "Bash",
"arguments": '{"command":"printf verified"}',
},
}
],
"workspace": str(tmp_path),
"session_id": "session-a",
}
class StubRunner:
def sign_observation(
self, session_id: str, payload: dict[str, object]
) -> dict[str, object]:
assert session_id == "session-a"
assert payload == expected_observation
return {**payload, "receipt_nonce": "nonce", "receipt_auth": "auth"}
def execute_exact_calls(
calls: tuple[ToolCall, ...],
*,
cwd: str,
timeout_s: float,
session_id: str,
) -> tuple[ToolExecutionResult, ...]:
assert len(calls) == 1
assert calls[0].name == "Bash"
assert calls[0].args == {"command": "printf verified"}
assert calls[0].call_id == "call-1"
assert calls[0].raw
assert cwd == str(tmp_path)
assert timeout_s == 0.0
assert session_id == "session-a"
return (executed,)
handler = object.__new__(Handler)
with (
mock.patch(
"nexum_runtime.server.execute_tool_calls",
side_effect=execute_exact_calls,
),
mock.patch("nexum_runtime.server._runner", return_value=StubRunner()),
mock.patch(
"nexum_runtime.server._json_response",
lambda _handler, status, payload: response.update(
status=status, payload=payload
),
),
):
handler.do_POST()
assert response == {
"status": 200,
"payload": {
"ok": True,
"results": [result_payload],
"observations": [
{
**expected_observation,
"receipt_nonce": "nonce",
"receipt_auth": "auth",
}
],
},
}
def test_server_rejects_signed_free_form_tool_execution(tmp_path: Path) -> None:
response: dict[str, object] = {}
class Handler(NexumHandler):
path = "/tools/execute"
model_dir = str(MODEL_DIR)
device = "cpu"
api_key = ""
enable_tools = True
workspace = str(tmp_path)
def _read_payload(self) -> dict[str, object]:
return {
"tool_text": "Bash(command='printf untrusted')",
"session_id": "session-a",
}
handler = object.__new__(Handler)
with mock.patch(
"nexum_runtime.server._json_response",
lambda _handler, status, payload: response.update(
status=status, payload=payload
),
):
handler.do_POST()
assert response == {
"status": 422,
"payload": {
"ok": False,
"error": (
"ValueError: signed execution requires exact structured tool_calls"
),
},
}
def test_server_does_not_expose_external_learning_ingestion() -> None:
response: dict[str, object] = {}
class Handler(NexumHandler):
path = "/external-learning"
model_dir = str(MODEL_DIR)
device = "cpu"
api_key = ""
enable_tools = True
def _read_payload(self) -> dict[str, object]:
return {
"session_id": "session-a",
"value": 1.0,
}
handler = object.__new__(Handler)
with mock.patch(
"nexum_runtime.server._json_response",
lambda _handler, status, payload: response.update(
status=status, payload=payload
),
):
handler.do_POST()
assert response == {
"status": 404,
"payload": {"ok": False, "error": "not_found"},
}
def test_server_attests_authenticated_external_observation() -> None:
response: dict[str, object] = {}
observation = {
"name": "Bash",
"args": {"command": "printf verified"},
"ok": True,
"output": "verified",
"executed": True,
"tool_call_id": "call-1",
}
caller_attested = {
**observation,
"receipt_source": "caller_attested",
"source_trust": "caller_owned",
}
class Handler(NexumHandler):
path = "/observations/attest"
model_dir = str(MODEL_DIR)
device = "cpu"
api_key = ""
enable_tools = True
def _read_payload(self) -> dict[str, object]:
return {"session_id": "session-a", "observation": observation}
class StubRunner:
def sign_observation(
self, session_id: str, payload: dict[str, object]
) -> dict[str, object]:
assert session_id == "session-a"
assert payload == caller_attested
return {**payload, "receipt_nonce": "nonce", "receipt_auth": "auth"}
handler = object.__new__(Handler)
with (
mock.patch("nexum_runtime.server._runner", return_value=StubRunner()),
mock.patch(
"nexum_runtime.server._json_response",
lambda _handler, status, payload: response.update(
status=status, payload=payload
),
),
):
handler.do_POST()
assert response == {
"status": 200,
"payload": {
"ok": True,
"observation": {
**caller_attested,
"receipt_nonce": "nonce",
"receipt_auth": "auth",
},
},
}