mirror of
https://github.com/turnstonelabs/turnstone.git
synced 2026-08-12 23:12:23 -06:00
Patrick Buckley
7a06f5e8bc
* refactor(session): make ModelLane the provider boundary (#979) ## Summary This closes the model-lane ownership gap left by #832: `ChatSession` no longer stores raw provider/client handles. `ResolvedModelBinding` now carries the provider, client, model, capabilities, registry generation, and backend-auth configuration as one coherent snapshot. - Atomically rebind existing sessions after model-registry changes while pinning each in-flight send, fallback, judge, output guard, task agent, title, compaction, perception, and voice operation to its initiating principal and binding. - Fence UI publication, canonical trajectory folds, durable writes, streams, retries, child scopes, and judge work by generation. Stop can hand off to a successor without accepting late state; cancelled tools retain typed effect receipts, and concurrent approval batches resolve by exact cycle or call. - Make create, fork, open, close, and delete race-safe with hidden `creating` reservations, incarnation-aware state tails, and an ACL-rechecked transaction that clones checkpoint-bounded history, configuration, project/persona state, and attachment references. - Extend REST/OpenAPI and Python/TypeScript SDK contracts for create/fork inputs, routed-create metadata, live-workstream probes, targeted approvals, and structured cancellation results. - Update architecture, storage, authentication, judge, channel, console, API, and SDK documentation, including regenerated architecture diagrams and OpenAPI artifacts. ## Validation - SQLite suite: 11,188 passed, 9 skipped, 10 deselected - PostgreSQL suite: 11,195 passed, 2 skipped, 10 deselected - Live backend: 3 passed - SSE recovery: 6 passed; browser recovery harness passed all scenarios - Ruff: clean; 595 files correctly formatted - mypy: 243 source files clean - TypeScript: typecheck/build and 35 tests passed - OpenAPI artifacts fresh; all 14 changed diagrams reproduce byte-for-byte - `git diff --check` and Git LFS integrity clean Closes #979. * fix(deps): update nanoid for GHSA-2v37-7h3g-55p8 Refresh the transitive lock entry admitted by PostCSS so the TypeScript security gate no longer resolves the vulnerable custom-generator implementation. Validation: - npm ci - npm audit --audit-level=moderate: 0 vulnerabilities - TypeScript typecheck and build - TypeScript tests: 35 passed * fix(test): assert canonical model registry URLs Replace prefix checks with exact canonical base URL assertions so the tests do not model incomplete URL validation. Validation: tests/test_model_registry.py (185 passed); Ruff check/format; mypy.
736 lines
26 KiB
Python
736 lines
26 KiB
Python
"""Backend-parity tests for the atomic workstream clone primitive.
|
|
|
|
The shared ``storage_backend`` fixture runs these against SQLite by default and
|
|
against PostgreSQL under ``--storage-backend=postgresql``. The contract lives at
|
|
the storage boundary: authorization, snapshot reads, destination writes, and
|
|
attachment retention either commit together or all roll back.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
|
|
import pytest
|
|
import sqlalchemy as sa
|
|
|
|
from turnstone.core.storage import (
|
|
ForkCloneExpectation,
|
|
ForkDestinationConflictError,
|
|
ForkSourceUnavailableError,
|
|
)
|
|
from turnstone.core.storage._protocol import FORK_RESERVATION_CONFIG_KEY
|
|
from turnstone.core.storage._schema import conversations, workstream_config, workstreams
|
|
from turnstone.core.trajectory import AttachmentRef, Role
|
|
|
|
|
|
def _register(
|
|
backend,
|
|
ws_id: str,
|
|
user_id: str,
|
|
*,
|
|
project_id: str | None = None,
|
|
fork_reservation_token: str = "",
|
|
state: str | None = None,
|
|
) -> None:
|
|
backend.register_workstream(
|
|
ws_id,
|
|
user_id=user_id,
|
|
project_id=project_id,
|
|
state=state or ("creating" if fork_reservation_token else "idle"),
|
|
kind="interactive",
|
|
fork_reservation_token=fork_reservation_token,
|
|
)
|
|
|
|
|
|
def _raw_conversation_rows(backend, ws_id: str) -> list[tuple[str, str | None, str | None]]:
|
|
with backend._conn() as conn:
|
|
rows = conn.execute(
|
|
sa.select(
|
|
conversations.c.content,
|
|
conversations.c.attachments,
|
|
conversations.c.meta,
|
|
)
|
|
.where(conversations.c.ws_id == ws_id)
|
|
.order_by(conversations.c.id)
|
|
).all()
|
|
return [(str(content or ""), attachments, meta) for content, attachments, meta in rows]
|
|
|
|
|
|
def _raw_workstream_config(backend, ws_id: str) -> dict[str, str]:
|
|
with backend._conn() as conn:
|
|
rows = conn.execute(
|
|
sa.select(workstream_config.c.key, workstream_config.c.value).where(
|
|
workstream_config.c.ws_id == ws_id
|
|
)
|
|
).all()
|
|
return {str(key): str(value) for key, value in rows}
|
|
|
|
|
|
def test_clone_accepts_empty_source_and_replaces_config_and_project(storage_backend) -> None:
|
|
backend = storage_backend
|
|
backend.create_project("shared", "Shared", "owner", visibility="public")
|
|
_register(backend, "source", "owner", project_id="shared")
|
|
_register(backend, "destination", "alice", project_id="shared", state="creating")
|
|
backend.save_workstream_config(
|
|
"source",
|
|
{"model_alias": "fast", "temperature": "0.25"},
|
|
)
|
|
backend.save_workstream_config("destination", {"stale": "yes"})
|
|
|
|
snapshot = backend.clone_workstream(
|
|
"source",
|
|
"destination",
|
|
principal_id="alice",
|
|
)
|
|
|
|
assert snapshot.turns == ()
|
|
assert snapshot.config == {"model_alias": "fast", "temperature": "0.25"}
|
|
assert snapshot.project_id == "shared"
|
|
assert backend.load_message_turns("destination") == []
|
|
assert backend.load_workstream_config("destination") == snapshot.config
|
|
destination = backend.get_workstream("destination")
|
|
assert destination is not None
|
|
assert destination["project_id"] == "shared"
|
|
|
|
|
|
@pytest.mark.parametrize("authorization_change", ["membership_revoked", "public_to_private"])
|
|
def test_clone_rechecks_current_project_authorization(
|
|
storage_backend,
|
|
authorization_change: str,
|
|
) -> None:
|
|
backend = storage_backend
|
|
visibility = "private" if authorization_change == "membership_revoked" else "public"
|
|
backend.create_project("project", "Project", "owner", visibility=visibility)
|
|
if authorization_change == "membership_revoked":
|
|
backend.add_project_member("project", "alice")
|
|
_register(backend, "source", "owner", project_id="project")
|
|
_register(backend, "destination", "alice", project_id="project", state="creating")
|
|
backend.save_message("source", "user", "private history")
|
|
backend.save_workstream_config("destination", {"keep": "unchanged"})
|
|
|
|
if authorization_change == "membership_revoked":
|
|
assert backend.remove_project_member("project", "alice") is True
|
|
else:
|
|
assert backend.update_project("project", visibility="private") is True
|
|
|
|
with pytest.raises(ForkSourceUnavailableError, match="source is no longer available"):
|
|
backend.clone_workstream("source", "destination", principal_id="alice")
|
|
|
|
assert backend.load_message_turns("destination") == []
|
|
assert backend.load_workstream_config("destination") == {"keep": "unchanged"}
|
|
|
|
|
|
@pytest.mark.parametrize("project_change", ["deleted", "rebound"])
|
|
def test_clone_refuses_source_project_change_after_destination_preflight(
|
|
storage_backend,
|
|
project_change: str,
|
|
) -> None:
|
|
backend = storage_backend
|
|
backend.create_project("original", "Original", "alice", visibility="public")
|
|
backend.create_project("replacement", "Replacement", "alice", visibility="public")
|
|
_register(backend, "source", "alice", project_id="original")
|
|
_register(backend, "destination", "alice", project_id="original", state="creating")
|
|
backend.save_message("source", "user", "must not copy")
|
|
backend.save_workstream_config("destination", {"keep": "unchanged"})
|
|
|
|
if project_change == "deleted":
|
|
assert backend.delete_project("original") is True
|
|
else:
|
|
with backend._conn() as conn:
|
|
conn.execute(
|
|
sa.update(workstreams)
|
|
.where(workstreams.c.ws_id == "source")
|
|
.values(project_id="replacement")
|
|
)
|
|
conn.commit()
|
|
|
|
with pytest.raises(ForkSourceUnavailableError, match="source project changed"):
|
|
backend.clone_workstream("source", "destination", principal_id="alice")
|
|
|
|
assert backend.load_message_turns("destination") == []
|
|
assert backend.load_workstream_config("destination") == {"keep": "unchanged"}
|
|
destination = backend.get_workstream("destination")
|
|
assert destination is not None
|
|
assert destination["project_id"] == "original"
|
|
|
|
|
|
def test_clone_rechecks_source_existence(storage_backend) -> None:
|
|
backend = storage_backend
|
|
_register(backend, "source", "alice")
|
|
_register(backend, "destination", "alice", state="creating")
|
|
backend.save_message("source", "user", "soon deleted")
|
|
assert backend.delete_workstream("source") is True
|
|
|
|
with pytest.raises(ForkSourceUnavailableError):
|
|
backend.clone_workstream("source", "destination", principal_id="alice")
|
|
|
|
assert backend.load_message_turns("destination") == []
|
|
|
|
|
|
def test_incarnation_snapshot_claims_legacy_token_without_public_exposure(storage_backend) -> None:
|
|
backend = storage_backend
|
|
_register(backend, "legacy", "alice")
|
|
|
|
first = backend.ensure_workstream_incarnation_snapshot("legacy")
|
|
second = backend.ensure_workstream_incarnation_snapshot("legacy")
|
|
|
|
assert first is not None and second is not None
|
|
token = first["fork_reservation_token"]
|
|
assert isinstance(token, str) and token
|
|
assert second["fork_reservation_token"] == token
|
|
backend.save_workstream_config(
|
|
"legacy",
|
|
{
|
|
"visible": "kept",
|
|
FORK_RESERVATION_CONFIG_KEY: "must-not-overwrite",
|
|
},
|
|
)
|
|
public_row = backend.get_workstream("legacy")
|
|
assert public_row is not None
|
|
assert "fork_reservation_token" not in public_row
|
|
assert backend.load_workstream_config("legacy") == {"visible": "kept"}
|
|
assert _raw_workstream_config(backend, "legacy") == {
|
|
FORK_RESERVATION_CONFIG_KEY: token,
|
|
"visible": "kept",
|
|
}
|
|
|
|
|
|
def test_clone_rejects_hidden_creating_source(storage_backend) -> None:
|
|
backend = storage_backend
|
|
_register(
|
|
backend,
|
|
"source",
|
|
"alice",
|
|
state="creating",
|
|
fork_reservation_token="source-incarnation",
|
|
)
|
|
_register(backend, "destination", "alice", state="creating")
|
|
backend.save_message("source", "user", "not yet published")
|
|
|
|
with pytest.raises(ForkSourceUnavailableError, match="source is no longer available"):
|
|
backend.clone_workstream("source", "destination", principal_id="alice")
|
|
|
|
assert backend.load_message_turns("destination") == []
|
|
|
|
|
|
def test_clone_refuses_same_id_source_replacement_after_preflight(storage_backend) -> None:
|
|
backend = storage_backend
|
|
_register(backend, "source", "alice")
|
|
backend.save_message("source", "user", "authorized predecessor")
|
|
source_snapshot = backend.ensure_workstream_incarnation_snapshot("source")
|
|
assert source_snapshot is not None
|
|
predecessor_token = source_snapshot["fork_reservation_token"]
|
|
|
|
assert backend.delete_workstream("source") is True
|
|
_register(
|
|
backend,
|
|
"source",
|
|
"alice",
|
|
state="idle",
|
|
fork_reservation_token="replacement-incarnation",
|
|
)
|
|
backend.save_message("source", "user", "replacement history")
|
|
_register(
|
|
backend,
|
|
"destination",
|
|
"alice",
|
|
fork_reservation_token="destination-incarnation",
|
|
)
|
|
expectation = ForkCloneExpectation(
|
|
persona_config=(),
|
|
project_id="",
|
|
project_name="",
|
|
project_writable=False,
|
|
destination_reservation_token="destination-incarnation",
|
|
source_reservation_token=predecessor_token,
|
|
)
|
|
|
|
with pytest.raises(ForkSourceUnavailableError, match="source is no longer available"):
|
|
backend.clone_workstream(
|
|
"source",
|
|
"destination",
|
|
principal_id="alice",
|
|
expected_session=expectation,
|
|
)
|
|
|
|
assert backend.load_message_turns("destination") == []
|
|
assert [turn.text for turn in backend.load_message_turns("source")] == ["replacement history"]
|
|
assert backend.get_workstream_reservation_token("source") == "replacement-incarnation"
|
|
|
|
|
|
def test_clone_refuses_nonempty_destination_without_mutation(storage_backend) -> None:
|
|
backend = storage_backend
|
|
_register(backend, "source", "alice")
|
|
_register(backend, "destination", "alice", state="creating")
|
|
backend.save_message("source", "user", "source")
|
|
backend.save_message("destination", "user", "existing")
|
|
backend.save_workstream_config("destination", {"keep": "yes"})
|
|
|
|
with pytest.raises(ForkDestinationConflictError, match="already has history"):
|
|
backend.clone_workstream("source", "destination", principal_id="alice")
|
|
|
|
assert [turn.text for turn in backend.load_message_turns("destination")] == ["existing"]
|
|
assert backend.load_workstream_config("destination") == {"keep": "yes"}
|
|
|
|
|
|
def test_clone_retains_matching_destination_reservation_privately(storage_backend) -> None:
|
|
backend = storage_backend
|
|
_register(backend, "source", "alice")
|
|
_register(
|
|
backend,
|
|
"destination",
|
|
"alice",
|
|
fork_reservation_token="destination-incarnation",
|
|
)
|
|
backend.save_message("source", "user", "copy me")
|
|
backend.save_workstream_config("source", {"source": "adopted"})
|
|
source_snapshot = backend.ensure_workstream_incarnation_snapshot("source")
|
|
assert source_snapshot is not None
|
|
|
|
snapshot = backend.clone_workstream(
|
|
"source",
|
|
"destination",
|
|
principal_id="alice",
|
|
expected_session=ForkCloneExpectation(
|
|
persona_config=(),
|
|
project_id="",
|
|
project_name="",
|
|
project_writable=False,
|
|
destination_reservation_token="destination-incarnation",
|
|
source_reservation_token=source_snapshot["fork_reservation_token"],
|
|
),
|
|
)
|
|
|
|
assert [turn.text for turn in snapshot.turns] == ["copy me"]
|
|
assert snapshot.config == {"source": "adopted"}
|
|
assert backend.load_workstream_config("destination") == snapshot.config
|
|
assert _raw_workstream_config(backend, "destination") == {
|
|
FORK_RESERVATION_CONFIG_KEY: "destination-incarnation",
|
|
"source": "adopted",
|
|
}
|
|
assert (
|
|
backend.delete_workstream_if_fork_reserved(
|
|
"destination",
|
|
"destination-incarnation",
|
|
)
|
|
is True
|
|
)
|
|
assert backend.get_workstream("destination") is None
|
|
|
|
|
|
def test_duplicate_registration_cannot_steal_destination_reservation(storage_backend) -> None:
|
|
backend = storage_backend
|
|
_register(
|
|
backend,
|
|
"destination",
|
|
"alice",
|
|
fork_reservation_token="incumbent",
|
|
)
|
|
|
|
inserted = backend.register_workstream(
|
|
"destination",
|
|
user_id="alice",
|
|
kind="interactive",
|
|
fork_reservation_token="challenger",
|
|
)
|
|
|
|
assert inserted is False
|
|
assert _raw_workstream_config(backend, "destination") == {
|
|
FORK_RESERVATION_CONFIG_KEY: "incumbent",
|
|
}
|
|
|
|
|
|
@pytest.mark.parametrize("new_token", ["", "fresh-incarnation"])
|
|
def test_registration_cannot_inherit_orphaned_reservation(
|
|
storage_backend,
|
|
new_token: str,
|
|
) -> None:
|
|
backend = storage_backend
|
|
with backend._conn() as conn:
|
|
conn.execute(
|
|
sa.insert(workstream_config),
|
|
{
|
|
"ws_id": "destination",
|
|
"key": FORK_RESERVATION_CONFIG_KEY,
|
|
"value": "orphaned-incarnation",
|
|
},
|
|
)
|
|
conn.commit()
|
|
|
|
_register(
|
|
backend,
|
|
"destination",
|
|
"alice",
|
|
fork_reservation_token=new_token,
|
|
)
|
|
|
|
expected = {FORK_RESERVATION_CONFIG_KEY: new_token} if new_token else {}
|
|
assert _raw_workstream_config(backend, "destination") == expected
|
|
assert (
|
|
backend.delete_workstream_if_fork_reserved(
|
|
"destination",
|
|
"orphaned-incarnation",
|
|
)
|
|
is False
|
|
)
|
|
assert backend.get_workstream("destination") is not None
|
|
|
|
|
|
def test_finalize_deferred_create_applies_all_writes_atomically(storage_backend) -> None:
|
|
backend = storage_backend
|
|
_register(
|
|
backend,
|
|
"destination",
|
|
"alice",
|
|
fork_reservation_token="destination-incarnation",
|
|
)
|
|
backend.save_workstream_config("destination", {"existing": "preserved"})
|
|
|
|
finalized = backend.finalize_deferred_create(
|
|
"destination",
|
|
"destination-incarnation",
|
|
alias="friendly-name",
|
|
config={
|
|
"new-setting": "installed",
|
|
FORK_RESERVATION_CONFIG_KEY: "must-not-overwrite",
|
|
},
|
|
node_id="node-a",
|
|
override_reason="local",
|
|
)
|
|
|
|
assert finalized is True
|
|
row = backend.get_workstream("destination")
|
|
assert row is not None
|
|
assert row["alias"] == "friendly-name"
|
|
assert backend.load_workstream_config("destination") == {
|
|
"existing": "preserved",
|
|
"new-setting": "installed",
|
|
}
|
|
assert _raw_workstream_config(backend, "destination") == {
|
|
FORK_RESERVATION_CONFIG_KEY: "destination-incarnation",
|
|
"existing": "preserved",
|
|
"new-setting": "installed",
|
|
}
|
|
overrides = backend.list_workstream_overrides()
|
|
assert len(overrides) == 1
|
|
assert overrides[0]["ws_id"] == "destination"
|
|
assert overrides[0]["node_id"] == "node-a"
|
|
assert overrides[0]["reason"] == "local"
|
|
|
|
|
|
def test_finalize_deferred_create_refuses_replaced_reservation(storage_backend) -> None:
|
|
backend = storage_backend
|
|
_register(
|
|
backend,
|
|
"destination",
|
|
"alice",
|
|
fork_reservation_token="first-incarnation",
|
|
)
|
|
assert backend.delete_workstream("destination") is True
|
|
_register(
|
|
backend,
|
|
"destination",
|
|
"alice",
|
|
fork_reservation_token="replacement-incarnation",
|
|
)
|
|
assert backend.set_workstream_alias("destination", "replacement-name") is True
|
|
backend.save_workstream_config("destination", {"replacement": "untouched"})
|
|
backend.set_workstream_override("destination", "node-b", reason="replacement")
|
|
|
|
finalized = backend.finalize_deferred_create(
|
|
"destination",
|
|
"first-incarnation",
|
|
alias="stale-name",
|
|
config={"stale": "must-not-land"},
|
|
node_id="node-a",
|
|
override_reason="local",
|
|
)
|
|
|
|
assert finalized is False
|
|
row = backend.get_workstream("destination")
|
|
assert row is not None
|
|
assert row["alias"] == "replacement-name"
|
|
assert backend.load_workstream_config("destination") == {"replacement": "untouched"}
|
|
assert _raw_workstream_config(backend, "destination") == {
|
|
FORK_RESERVATION_CONFIG_KEY: "replacement-incarnation",
|
|
"replacement": "untouched",
|
|
}
|
|
overrides = backend.list_workstream_overrides()
|
|
assert len(overrides) == 1
|
|
assert overrides[0]["ws_id"] == "destination"
|
|
assert overrides[0]["node_id"] == "node-b"
|
|
assert overrides[0]["reason"] == "replacement"
|
|
|
|
|
|
def test_finalize_deferred_create_alias_conflict_rolls_back_other_writes(
|
|
storage_backend,
|
|
) -> None:
|
|
backend = storage_backend
|
|
_register(backend, "incumbent", "alice")
|
|
assert backend.set_workstream_alias("incumbent", "taken-name") is True
|
|
_register(
|
|
backend,
|
|
"destination",
|
|
"alice",
|
|
fork_reservation_token="destination-incarnation",
|
|
)
|
|
backend.save_workstream_config("destination", {"existing": "preserved"})
|
|
backend.set_workstream_override("destination", "node-before", reason="existing")
|
|
|
|
finalized = backend.finalize_deferred_create(
|
|
"destination",
|
|
"destination-incarnation",
|
|
alias="taken-name",
|
|
config={"stale": "must-not-land"},
|
|
node_id="node-after",
|
|
override_reason="local",
|
|
)
|
|
|
|
assert finalized is False
|
|
row = backend.get_workstream("destination")
|
|
assert row is not None
|
|
assert row["alias"] is None
|
|
assert backend.load_workstream_config("destination") == {"existing": "preserved"}
|
|
overrides = backend.list_workstream_overrides()
|
|
destination_override = next(row for row in overrides if row["ws_id"] == "destination")
|
|
assert destination_override["node_id"] == "node-before"
|
|
assert destination_override["reason"] == "existing"
|
|
|
|
|
|
def test_clone_refuses_replaced_destination_reservation(storage_backend) -> None:
|
|
"""A same-id replacement cannot inherit an earlier create's clone."""
|
|
backend = storage_backend
|
|
_register(backend, "source", "alice")
|
|
_register(
|
|
backend,
|
|
"destination",
|
|
"alice",
|
|
fork_reservation_token="first-incarnation",
|
|
)
|
|
backend.save_message("source", "user", "must not copy")
|
|
backend.save_workstream_config("source", {"source": "unchanged"})
|
|
source_snapshot = backend.ensure_workstream_incarnation_snapshot("source")
|
|
assert source_snapshot is not None
|
|
|
|
assert backend.delete_workstream("destination") is True
|
|
_register(
|
|
backend,
|
|
"destination",
|
|
"alice",
|
|
fork_reservation_token="replacement-incarnation",
|
|
)
|
|
backend.save_workstream_config("destination", {"replacement": "untouched"})
|
|
|
|
expectation = ForkCloneExpectation(
|
|
persona_config=(),
|
|
project_id="",
|
|
project_name="",
|
|
project_writable=False,
|
|
destination_reservation_token="first-incarnation",
|
|
source_reservation_token=source_snapshot["fork_reservation_token"],
|
|
)
|
|
with pytest.raises(ForkDestinationConflictError, match="destination is not available"):
|
|
backend.clone_workstream(
|
|
"source",
|
|
"destination",
|
|
principal_id="alice",
|
|
expected_session=expectation,
|
|
)
|
|
|
|
assert backend.load_message_turns("destination") == []
|
|
assert backend.load_workstream_config("destination") == {
|
|
"replacement": "untouched",
|
|
}
|
|
assert _raw_workstream_config(backend, "destination") == {
|
|
FORK_RESERVATION_CONFIG_KEY: "replacement-incarnation",
|
|
"replacement": "untouched",
|
|
}
|
|
assert (
|
|
backend.delete_workstream_if_fork_reserved(
|
|
"destination",
|
|
"first-incarnation",
|
|
)
|
|
is False
|
|
)
|
|
assert [turn.text for turn in backend.load_message_turns("source")] == ["must not copy"]
|
|
assert backend.load_workstream_config("source") == {"source": "unchanged"}
|
|
|
|
|
|
def test_clone_preserves_raw_attachment_refs_and_balances_refcounts(storage_backend) -> None:
|
|
backend = storage_backend
|
|
_register(backend, "source", "alice")
|
|
destination_token = "attachment-clone-incarnation"
|
|
_register(
|
|
backend,
|
|
"destination",
|
|
"alice",
|
|
state="creating",
|
|
fork_reservation_token=destination_token,
|
|
)
|
|
|
|
document_id = "a" * 64
|
|
preview_id = "b" * 64
|
|
user_row = backend.save_message("source", "user", "read this")
|
|
backend.save_attachment(
|
|
document_id,
|
|
"notes.txt",
|
|
"text/plain",
|
|
5,
|
|
"text",
|
|
b"notes",
|
|
)
|
|
backend.set_message_attachments("source", user_row, [document_id])
|
|
|
|
tool_calls = json.dumps(
|
|
[
|
|
{
|
|
"id": "call-1",
|
|
"type": "function",
|
|
"function": {"name": "render", "arguments": "{}"},
|
|
}
|
|
]
|
|
)
|
|
backend.save_message("source", "assistant", None, tool_calls=tool_calls)
|
|
preview_meta = {
|
|
"effect_status": "committed",
|
|
"preview": {"attachment_id": preview_id, "title": "Rendered output"},
|
|
}
|
|
tool_row = backend.save_message(
|
|
"source",
|
|
"tool",
|
|
"rendered",
|
|
tool_call_id="call-1",
|
|
meta=json.dumps(preview_meta),
|
|
)
|
|
backend.save_attachment(
|
|
preview_id,
|
|
"preview.html",
|
|
"text/html",
|
|
7,
|
|
"preview",
|
|
b"preview",
|
|
origin="tool",
|
|
)
|
|
backend.set_message_attachments("source", tool_row, [preview_id])
|
|
backend.save_message("source", "assistant", "done")
|
|
backend.save_workstream_config("destination", {"stale": "remove-me"})
|
|
|
|
snapshot = backend.clone_workstream(
|
|
"source",
|
|
"destination",
|
|
principal_id="alice",
|
|
)
|
|
|
|
assert [turn.text for turn in snapshot.turns] == ["read this", "", "rendered", "done"]
|
|
user_turn = snapshot.turns[0]
|
|
assert user_turn.role is Role.USER
|
|
assert [
|
|
block.attachment_id for block in user_turn.content if isinstance(block, AttachmentRef)
|
|
] == [document_id]
|
|
tool_turn = snapshot.turns[2]
|
|
assert tool_turn.meta.extra["preview"]["attachment_id"] == preview_id
|
|
assert tool_turn.meta.extra["storage_attachment_ids"] == [preview_id]
|
|
assert backend.load_workstream_config("destination") == {}
|
|
|
|
raw_rows = _raw_conversation_rows(backend, "destination")
|
|
assert json.loads(raw_rows[0][1] or "[]") == [document_id]
|
|
assert json.loads(raw_rows[2][1] or "[]") == [preview_id]
|
|
assert json.loads(raw_rows[2][2] or "{}") == preview_meta
|
|
# The clone transaction commits before lifecycle publication, but ordinary
|
|
# recall must not expose that provisional transcript. Exact publication
|
|
# flips the same durable incarnation to visible.
|
|
assert not any(row[1] == "destination" for row in backend.search_history("rendered"))
|
|
assert backend.publish_deferred_create("destination", destination_token) is True
|
|
assert any(row[1] == "destination" for row in backend.search_history("rendered"))
|
|
for attachment_id in (document_id, preview_id):
|
|
attachment = backend.get_attachment(attachment_id)
|
|
assert attachment is not None
|
|
assert attachment["refcount"] == 2
|
|
|
|
assert backend.delete_workstream("source") is True
|
|
for attachment_id in (document_id, preview_id):
|
|
attachment = backend.get_attachment(attachment_id)
|
|
assert attachment is not None
|
|
assert attachment["refcount"] == 1
|
|
|
|
|
|
def test_missing_attachment_rolls_back_refs_config_history_and_binding(storage_backend) -> None:
|
|
backend = storage_backend
|
|
backend.create_project("old-project", "Old", "alice", visibility="private")
|
|
_register(backend, "source", "alice", project_id="old-project")
|
|
_register(
|
|
backend,
|
|
"destination",
|
|
"alice",
|
|
project_id="old-project",
|
|
state="creating",
|
|
)
|
|
existing_id = "c" * 64
|
|
missing_id = "d" * 64
|
|
source_row = backend.save_message("source", "user", "two refs")
|
|
backend.save_attachment(
|
|
existing_id,
|
|
"exists.txt",
|
|
"text/plain",
|
|
6,
|
|
"text",
|
|
b"exists",
|
|
)
|
|
backend.set_message_attachments("source", source_row, [existing_id, missing_id])
|
|
backend.save_workstream_config("source", {"source": "value"})
|
|
backend.save_workstream_config("destination", {"keep": "value"})
|
|
|
|
with pytest.raises(ForkSourceUnavailableError, match="attachments are no longer available"):
|
|
backend.clone_workstream("source", "destination", principal_id="alice")
|
|
|
|
existing = backend.get_attachment(existing_id)
|
|
assert existing is not None
|
|
assert existing["refcount"] == 1
|
|
assert backend.load_message_turns("destination") == []
|
|
assert backend.load_workstream_config("destination") == {"keep": "value"}
|
|
destination = backend.get_workstream("destination")
|
|
assert destination is not None
|
|
assert destination["project_id"] == "old-project"
|
|
|
|
|
|
def test_compacted_clone_rewrites_marker_to_destination_id_space(storage_backend) -> None:
|
|
backend = storage_backend
|
|
_register(backend, "source", "alice")
|
|
_register(backend, "destination", "alice", state="creating")
|
|
backend.save_message("source", "user", "old question")
|
|
backend.save_message("source", "assistant", "old answer")
|
|
source_watermark = backend.get_compaction_watermark("source")
|
|
assert source_watermark is not None
|
|
backend.save_message(
|
|
"source",
|
|
"assistant",
|
|
"SUMMARY",
|
|
source="compaction",
|
|
meta=json.dumps({"watermark": source_watermark, "input_tokens": 321}),
|
|
)
|
|
backend.save_message("source", "user", "new question")
|
|
backend.save_message("source", "assistant", "new answer")
|
|
|
|
snapshot = backend.clone_workstream(
|
|
"source",
|
|
"destination",
|
|
principal_id="alice",
|
|
)
|
|
|
|
expected = ["[Conversation summary]", "SUMMARY", "new question", "new answer"]
|
|
assert [turn.text for turn in snapshot.turns] == expected
|
|
assert [turn.text for turn in backend.load_message_turns("destination")] == expected
|
|
with backend._conn() as conn:
|
|
marker = conn.execute(
|
|
sa.select(conversations.c.id, conversations.c.meta).where(
|
|
conversations.c.ws_id == "destination",
|
|
conversations.c._source == "compaction",
|
|
)
|
|
).one()
|
|
marker_id = int(marker[0])
|
|
marker_meta = json.loads(marker[1])
|
|
assert marker_meta == {"watermark": marker_id, "input_tokens": 321}
|
|
assert marker_id != source_watermark
|
|
assert backend.get_compaction_checkpoint("destination") == marker_id
|
|
assert backend.count_messages("destination") == 3 # marker + two live tail rows
|