"""Direct contracts for interactive artifact mutations.""" from __future__ import annotations import base64 import copy import hashlib from pathlib import Path import pytest from openai4s.config import Config, LLMConfig from openai4s.server import artifacts as artifacts_mod from openai4s.server.artifacts import ArtifactManager, ArtifactOperationError from openai4s.store import get_store class MutationHarness: def __init__(self, tmp_path: Path) -> None: self.cfg = Config( data_dir=tmp_path / "data", llm=LLMConfig(provider="deepseek", api_key="test-key"), ) self.store = get_store(self.cfg.db_path) self.frame_id = self.store.new_frame( kind="turn", project_id="default", status="ready" ) self.workspace = self.cfg.data_dir / "workspaces" / self.frame_id self.workspace.mkdir(parents=True) self.events: list[tuple[str, dict]] = [] self.manager = ArtifactManager( data_dir=self.cfg.data_dir, store=self.store, workspace_for=lambda frame_id: self.workspace, broadcast=lambda frame_id, event: self.events.append((frame_id, event)), guess_content_type=lambda name: ( "text/plain; charset=utf-8" if name.endswith(".txt") else "application/octet-stream" ), checksum=lambda path: hashlib.sha256(path.read_bytes()).hexdigest(), ) def artifact(self, filename: str, data: bytes, content_type: str) -> dict: path = self.workspace / filename path.parent.mkdir(parents=True, exist_ok=True) path.write_bytes(data) return self.store.save_artifact( path=str(path), filename=filename, content_type=content_type, size_bytes=len(data), checksum=hashlib.sha256(data).hexdigest(), frame_id=self.frame_id, project_id="default", ) def raised_operation(call, code: int, message: str) -> None: with pytest.raises(ArtifactOperationError) as caught: call() assert caught.value.code == code assert caught.value.message == message assert str(caught.value) == message def test_edit_versions_live_text_and_preserves_exact_event_shape(tmp_path): harness = MutationHarness(tmp_path) first = harness.artifact("notes.txt", b"version one", "text/plain") artifact_id = first["artifact_id"] override: list[tuple[str, dict]] = [] result = harness.manager.edit( artifact_id, "version two", broadcast=lambda frame_id, event: override.append((frame_id, event)), ) assert result == { "ok": True, "artifact_id": artifact_id, "version_id": result["version_id"], "size_bytes": len(b"version two"), "unchanged": False, } assert (harness.workspace / "notes.txt").read_text() == "version two" assert ( Path( harness.store.version_meta(first["version_id"])["snapshot_path"] ).read_bytes() == b"version one" ) assert ( Path( harness.store.version_meta(result["version_id"])["snapshot_path"] ).read_bytes() == b"version two" ) assert len(harness.store.list_versions(artifact_id)) == 2 assert harness.events == [] assert override == [ ( harness.frame_id, { "type": "artifact_created", "artifact": { "id": artifact_id, "filename": "notes.txt", "version_id": result["version_id"], "root_frame_id": harness.frame_id, }, }, ) ] def test_edit_rejects_missing_binary_and_write_failure(tmp_path, monkeypatch): harness = MutationHarness(tmp_path) raised_operation( lambda: harness.manager.edit("missing", "content"), 404, "artifact not found", ) image = harness.artifact("figure.png", b"PNG", "image/png") raised_operation( lambda: harness.manager.edit(image["artifact_id"], "content"), 415, "artifact is not text-editable", ) text = harness.artifact("write.txt", b"old", "text/plain") canary = "/Users/canary/Documents/embargoed.csv" def fail_write(_pinned, _data): raise OSError(28, "No space left on device", canary) monkeypatch.setattr(artifacts_mod._PinnedUploadFile, "write", fail_write) # The `strerror` no longer travels. A real `OSError` from this write carries # the absolute path it failed on -- under the data directory, so the # account's username -- and a 500 body is a public surface. The original # goes to the operator diagnostic instead. raised_operation( lambda: harness.manager.edit(text["artifact_id"], "new"), 500, "write failed", ) with pytest.raises(ArtifactOperationError) as caught: harness.manager.edit(text["artifact_id"], "new") assert canary not in str(caught.value) assert canary not in repr(caught.value.__dict__) def test_nested_edit_keeps_the_exact_artifact_path(tmp_path): harness = MutationHarness(tmp_path) first = harness.artifact("results/notes.txt", b"old\n", "text/plain") unchanged = harness.manager.edit(first["artifact_id"], "old\n") assert unchanged["unchanged"] is True assert unchanged["version_id"] == first["version_id"] assert ( Path( harness.store.version_meta(first["version_id"])["snapshot_path"] ).read_bytes() == b"old\n" ) edited = harness.manager.edit(first["artifact_id"], "new\n") target = harness.workspace / "results" / "notes.txt" assert target.read_bytes() == b"new\n" stored = harness.store.get_artifact(first["artifact_id"]) assert stored["filename"] == "results/notes.txt" assert stored["latest_version_id"] == edited["version_id"] versions = harness.store.list_versions(first["artifact_id"]) assert len(versions) == 2 assert { Path( harness.store.version_meta(version["version_id"])["snapshot_path"] ).read_bytes() for version in versions } == {b"old\n", b"new\n"} def test_nested_edit_parent_swap_cannot_redirect_the_daemon(tmp_path, monkeypatch): harness = MutationHarness(tmp_path) first = harness.artifact("results/notes.txt", b"old\n", "text/plain") harness.manager.write_version_snapshot( first["version_id"], first["filename"], data=b"old\n" ) before = copy.deepcopy(harness.store.get_artifact(first["artifact_id"])) before_versions = copy.deepcopy(harness.store.list_versions(first["artifact_id"])) parent = harness.workspace / "results" parked = harness.workspace / "parked-results" outside = tmp_path / "outside" outside.mkdir() outside_target = outside / "notes.txt" outside_target.write_bytes(b"OUTSIDE_CANARY\n") real_write = artifacts_mod._PinnedUploadFile.write swapped = False def swap_parent_then_write(pinned, data): nonlocal swapped if not swapped: parent.rename(parked) try: parent.symlink_to(outside, target_is_directory=True) except (OSError, NotImplementedError): parked.rename(parent) pytest.skip("directory symlinks are unavailable") swapped = True return real_write(pinned, data) monkeypatch.setattr( artifacts_mod._PinnedUploadFile, "write", swap_parent_then_write ) raised_operation( lambda: harness.manager.edit(first["artifact_id"], "new\n"), 500, "write failed", ) assert swapped is True assert outside_target.read_bytes() == b"OUTSIDE_CANARY\n" assert (parked / "notes.txt").read_bytes() == b"old\n" assert harness.store.get_artifact(first["artifact_id"]) == before assert harness.store.list_versions(first["artifact_id"]) == before_versions assert not list(parked.glob("*.part")) assert not list(harness.manager.versions_dir().glob(".pending-*")) assert not list(harness.manager.versions_dir().glob(".upload-*.json")) def test_nested_edit_rejects_a_swapped_workspace_root(tmp_path): harness = MutationHarness(tmp_path) first = harness.artifact("results/notes.txt", b"old\n", "text/plain") harness.manager.write_version_snapshot( first["version_id"], first["filename"], data=b"old\n" ) before = copy.deepcopy(harness.store.get_artifact(first["artifact_id"])) parked = harness.workspace.with_name("parked-workspace") outside = tmp_path / "outside-workspace" outside_target = outside / "results" / "notes.txt" outside_target.parent.mkdir(parents=True) outside_target.write_bytes(b"OUTSIDE_CANARY\n") harness.workspace.rename(parked) try: harness.workspace.symlink_to(outside, target_is_directory=True) except (OSError, NotImplementedError): parked.rename(harness.workspace) pytest.skip("directory symlinks are unavailable") raised_operation( lambda: harness.manager.edit(first["artifact_id"], "new\n"), 500, "write failed", ) assert outside_target.read_bytes() == b"OUTSIDE_CANARY\n" assert (parked / "results" / "notes.txt").read_bytes() == b"old\n" assert harness.store.get_artifact(first["artifact_id"]) == before assert len(harness.store.list_versions(first["artifact_id"])) == 1 assert not list(harness.manager.versions_dir().glob(".pending-*")) assert not list(harness.manager.versions_dir().glob(".upload-*.json")) @pytest.mark.parametrize("alias_kind", ["symlink", "hardlink"]) @pytest.mark.parametrize( "replacement", ["old\n", "new\n"], ids=["unchanged", "changed"] ) def test_nested_edit_rejects_final_component_aliases(tmp_path, alias_kind, replacement): harness = MutationHarness(tmp_path) first = harness.artifact("results/notes.txt", b"old\n", "text/plain") harness.manager.write_version_snapshot( first["version_id"], first["filename"], data=b"old\n" ) before = copy.deepcopy(harness.store.get_artifact(first["artifact_id"])) before_versions = copy.deepcopy(harness.store.list_versions(first["artifact_id"])) target = harness.workspace / "results" / "notes.txt" outside = tmp_path / "outside.txt" outside.write_bytes(b"OUTSIDE_CANARY\n") target.unlink() try: if alias_kind == "symlink": target.symlink_to(outside) else: artifacts_mod.os.link(outside, target) except (OSError, NotImplementedError): pytest.skip(f"{alias_kind} creation is unavailable") raised_operation( lambda: harness.manager.edit(first["artifact_id"], replacement), 500, "write failed", ) assert outside.read_bytes() == b"OUTSIDE_CANARY\n" assert harness.store.get_artifact(first["artifact_id"]) == before assert harness.store.list_versions(first["artifact_id"]) == before_versions assert not list(target.parent.glob("*.part")) assert not list(harness.manager.versions_dir().glob(".pending-*")) assert not list(harness.manager.versions_dir().glob(".upload-*.json")) def test_nested_edit_publish_failure_restores_every_prior_truth(tmp_path, monkeypatch): harness = MutationHarness(tmp_path) first = harness.artifact("results/notes.txt", b"old\n", "text/plain") harness.manager.write_version_snapshot( first["version_id"], first["filename"], data=b"old\n" ) before = copy.deepcopy(harness.store.get_artifact(first["artifact_id"])) before_versions = copy.deepcopy(harness.store.list_versions(first["artifact_id"])) real_replace = artifacts_mod._PinnedUploadDirectory.replace def fail_live_publish(directory, source, destination): if source.name.endswith(".part") and destination.name == "notes.txt": raise OSError("injected exact-Artifact publish failure") return real_replace(directory, source, destination) monkeypatch.setattr( artifacts_mod._PinnedUploadDirectory, "replace", fail_live_publish ) raised_operation( lambda: harness.manager.edit(first["artifact_id"], "new\n"), 500, "write failed", ) target = harness.workspace / "results" / "notes.txt" assert target.read_bytes() == b"old\n" assert harness.store.get_artifact(first["artifact_id"]) == before assert harness.store.list_versions(first["artifact_id"]) == before_versions assert not list(target.parent.glob("*.part")) assert not list(target.parent.glob(".*.backup")) assert not list(harness.manager.versions_dir().glob(".pending-*")) assert not list(harness.manager.versions_dir().glob(".upload-*.json")) def test_nested_edit_fails_closed_without_dirfd_capabilities(tmp_path, monkeypatch): harness = MutationHarness(tmp_path) first = harness.artifact("results/notes.txt", b"old\n", "text/plain") monkeypatch.setattr(artifacts_mod.os, "supports_dir_fd", set()) raised_operation( lambda: harness.manager.edit(first["artifact_id"], "new\n"), 500, "write failed", ) assert (harness.workspace / "results" / "notes.txt").read_bytes() == b"old\n" assert len(harness.store.list_versions(first["artifact_id"])) == 1 def test_log_extension_preserves_legacy_text_editability(tmp_path): harness = MutationHarness(tmp_path) log = harness.artifact("run.log", b"old", "application/octet-stream") result = harness.manager.edit(log["artifact_id"], "new") assert result["ok"] is True assert (harness.workspace / "run.log").read_text() == "new" def test_rename_changes_metadata_only_and_preserves_exact_event_shape(tmp_path): harness = MutationHarness(tmp_path) record = harness.artifact("before.txt", b"science", "text/plain") artifact_id = record["artifact_id"] result = harness.manager.rename(artifact_id, "after.txt") assert result == { "ok": True, "artifact_id": artifact_id, "filename": "after.txt", } assert harness.store.get_artifact(artifact_id)["filename"] == "after.txt" assert (harness.workspace / "before.txt").read_bytes() == b"science" assert not (harness.workspace / "after.txt").exists() assert harness.events == [ ( harness.frame_id, { "type": "artifact_created", "artifact": { "id": artifact_id, "filename": "after.txt", "root_frame_id": harness.frame_id, }, }, ) ] raised_operation( lambda: harness.manager.rename("missing", None), 400, "filename required", ) raised_operation( lambda: harness.manager.rename("missing", "new.txt"), 404, "artifact not found", ) def test_artifact_mutations_fail_closed_on_workspace_escape_metadata(tmp_path): harness = MutationHarness(tmp_path) record = harness.artifact("safe.txt", b"safe", "text/plain") outside = tmp_path / "outside.txt" outside.write_text("sentinel", encoding="utf-8") raised_operation( lambda: harness.manager.rename(record["artifact_id"], "../../outside.txt"), 400, "artifact live path escapes its workspace", ) assert harness.store.get_artifact(record["artifact_id"])["filename"] == "safe.txt" # Exact edits derive their live authority from the current version row, # not mutable presentation metadata. Corrupting the latter therefore # cannot redirect the writer outside the workspace. harness.store.rename_artifact(record["artifact_id"], "../../outside.txt") edited = harness.manager.edit(record["artifact_id"], "compromised") assert edited["unchanged"] is False assert (harness.workspace / "safe.txt").read_text("utf-8") == "compromised" assert outside.read_text("utf-8") == "sentinel" def test_upload_versioning_and_event_contracts(tmp_path): harness = MutationHarness(tmp_path) first = harness.manager.upload( { "filename": "../result.txt", "content_base64": base64.b64encode(b"alpha").decode(), "frame_id": harness.frame_id, } ) # Line-wrapped base64 is transport formatting and still decodes. The # payload here used to be "YmV0YQ==!", whose trailing "!" was silently # discarded -- this test pinned that leniency, and leniency is what makes # a corrupted upload undetectable. second = harness.manager.upload( { "filename": "result.txt", "content_base64": "YmV0\nYQ==\n", "frame_id": harness.frame_id, } ) assert first["artifact_id"] == first["id"] == second["artifact_id"] assert first["filename"] == second["filename"] == "result.txt" versions = harness.store.list_versions(first["artifact_id"]) assert len(versions) == 2 by_ordinal = {version["ordinal"]: version for version in versions} assert ( Path( harness.store.resolve_artifact_path(by_ordinal[1]["version_id"]) ).read_bytes() == b"alpha" ) assert ( Path( harness.store.resolve_artifact_path(by_ordinal[2]["version_id"]) ).read_bytes() == b"beta" ) assert harness.events[-1] == ( harness.frame_id, { "type": "artifact_created", "artifact": { "id": first["artifact_id"], "filename": "result.txt", "content_type": "text/plain; charset=utf-8", "root_frame_id": harness.frame_id, }, }, ) # This used to succeed, storing the literal text as the file's bytes and # hashing that. Upload a `.npy` whose payload lost a character in transit # and the artifact carried the base64 string instead of the array -- # versioned, checksummed and indistinguishable from data. A caller who # means to upload text says `content_text`. with pytest.raises(ArtifactOperationError) as refused: harness.manager.upload( { "filename": "fallback.bin", "content_base64": "%%% not base64 %%%", "frame_id": harness.frame_id, } ) assert refused.value.code == 400 assert "not valid base64" in refused.value.message # Nothing was written: a rejected upload leaves no artifact behind. assert not ( harness.manager.workspace_for(harness.frame_id) / "fallback.bin" ).exists() event_count = len(harness.events) loose = harness.manager.upload( { "filename": "loose.bin", "content_base64": base64.b64encode(b"outside").decode(), } ) assert len(harness.events) == event_count assert (harness.cfg.data_dir / "uploads" / "loose.bin").read_bytes() == b"outside" assert harness.store.get_artifact(loose["artifact_id"])["root_frame_id"] is None def test_delete_reclaims_versions_and_emits_bare_refresh_event(tmp_path): harness = MutationHarness(tmp_path) first = harness.artifact("delete.txt", b"one", "text/plain") artifact_id = first["artifact_id"] second = harness.manager.edit(artifact_id, "two") version_paths = [] for version_id in (first["version_id"], second["version_id"]): metadata = harness.store.version_meta(version_id) version_paths.extend([metadata["path"], metadata["snapshot_path"]]) harness.events.clear() assert harness.manager.delete(artifact_id) == {"ok": True} assert harness.store.get_artifact(artifact_id) is None assert all(not Path(path).exists() for path in set(version_paths)) assert harness.events == [ ( harness.frame_id, { "type": "artifact_created", "root_frame_id": harness.frame_id, }, ) ] harness.events.clear() assert harness.manager.delete("missing") == {"ok": True} assert harness.events == [] def test_upload_refuses_corrupted_base64_instead_of_decoding_it_to_other_bytes( tmp_path, ): """`b64decode` without `validate=True` silently discards stray characters. That is the half that never raised. A payload corrupted in transit decodes to *different bytes* and reports success, so the artifact carries a checksum computed over content nobody sent -- provenance that is wrong rather than absent, which is worse because it is believed. Whitespace is exempt because it is transport formatting: plenty of tools wrap base64 at 76 columns, and rejecting that would break honest callers while catching nothing. """ harness = MutationHarness(tmp_path) payload = base64.b64encode(b"\x00\x01measurement\xff").decode() clean = harness.manager.upload( { "filename": "clean.bin", "content_base64": payload, "frame_id": harness.frame_id, } ) stored = Path( harness.store.resolve_artifact_path(clean["artifact_id"]) ).read_bytes() assert stored == b"\x00\x01measurement\xff" # Wrapped at 76 columns: still the same bytes. wrapped = "\n".join(payload[i : i + 4] for i in range(0, len(payload), 4)) ok = harness.manager.upload( { "filename": "wrapped.bin", "content_base64": wrapped, "frame_id": harness.frame_id, } ) assert ( Path(harness.store.resolve_artifact_path(ok["artifact_id"])).read_bytes() == b"\x00\x01measurement\xff" ) # A stray non-alphabet character is corruption, not formatting. for corrupt in (payload[:4] + "!" + payload[4:], payload[:4] + "*" + payload[4:]): with pytest.raises(ArtifactOperationError) as refused: harness.manager.upload( { "filename": "corrupt.bin", "content_base64": corrupt, "frame_id": harness.frame_id, } ) assert refused.value.code == 400 def test_upload_will_not_guess_which_content_field_is_authoritative(tmp_path): """Three fields, one meaning. Two supplied is a question, not a payload.""" harness = MutationHarness(tmp_path) with pytest.raises(ArtifactOperationError) as refused: harness.manager.upload( { "filename": "both.txt", "content_base64": base64.b64encode(b"a").decode(), "content_text": "b", "frame_id": harness.frame_id, } ) assert refused.value.code == 400 assert "exactly one" in refused.value.message # And content_text is the sanctioned way to upload text. text = harness.manager.upload( { "filename": "note.txt", "content_text": "plain text, not base64", "frame_id": harness.frame_id, } ) assert ( Path(harness.store.resolve_artifact_path(text["artifact_id"])).read_bytes() == b"plain text, not base64" )