From 56f0a3f52c5850431848c198fd4aa88f696516c2 Mon Sep 17 00:00:00 2001 From: witbrock Date: Wed, 16 Sep 2026 10:43:11 -0400 Subject: [PATCH 1/3] Add fenced task-project homes and signed snapshot capture --- .../services/task_management_service.py | 33 ++ .../services/task_project_home_service.py | 306 ++++++++++++++++++ .../services/task_project_transfer_service.py | 272 ++++++++++++++++ .../backend/test_task_project_home_service.py | 110 +++++++ .../test_task_project_transfer_service.py | 156 +++++++++ 5 files changed, 877 insertions(+) create mode 100644 src/backend/services/task_project_home_service.py create mode 100644 src/backend/services/task_project_transfer_service.py create mode 100644 tests/backend/test_task_project_home_service.py create mode 100644 tests/backend/test_task_project_transfer_service.py diff --git a/src/backend/services/task_management_service.py b/src/backend/services/task_management_service.py index 822cca63..3b121eb6 100644 --- a/src/backend/services/task_management_service.py +++ b/src/backend/services/task_management_service.py @@ -19,6 +19,7 @@ from ..db.repositories.concepts_repository import ConceptsRepository from ..security.access_control import can_access_concept +from .task_project_home_service import guard_task_write from ..security.visibility_predicates import ( CANONICAL_SPECIFIC_TO_ORG_PREDICATE, SPECIFIC_TO_ORG_PREDICATES_READ, @@ -1307,6 +1308,7 @@ def _would_create_parent_cycle( return False +@guard_task_write def create_task( title: str, description: str, @@ -2265,6 +2267,18 @@ def update_task_status( f"Invalid status '{status}'. Must be one of: {VALID_TASK_STATUSES}" ) + return _update_validated_task_status( + task_concept_id, status, actor_concept_id=actor_concept_id + ) + + +@guard_task_write +def _update_validated_task_status( + task_concept_id: str, + status: str, + *, + actor_concept_id: str | None = None, +) -> Dict[str, Any]: task_concept_id, doc = _get_task_doc(task_concept_id) existing_task = _build_task_response(doc) @@ -2397,6 +2411,7 @@ def update_task_status( return updated_task +@guard_task_write def assign_task( task_concept_id: str, assignee_concept_id: str, *, update_visibility: bool = True ) -> Dict[str, Any]: @@ -3940,6 +3955,7 @@ def get_task_transitions(task_concept_id: str) -> Dict[str, Any]: } +@guard_task_write def transition_task( task_concept_id: str, *, @@ -4011,6 +4027,7 @@ def transition_task( } +@guard_task_write def unassign_task(task_concept_id: str) -> Dict[str, Any]: task_concept_id, doc = _get_task_doc(task_concept_id) current_assignees = doc.get("relationships", {}).get(PREDICATE_HAS_ASSIGNEE, []) @@ -4050,6 +4067,7 @@ def unassign_task(task_concept_id: str) -> Dict[str, Any]: return get_task(task_concept_id) +@guard_task_write def set_task_parent( task_concept_id: str, parent_task_concept_id: str | None, @@ -4130,6 +4148,7 @@ def set_task_parent( return get_task(task_concept_id) +@guard_task_write def create_subtask( parent_task_concept_id: str, *, @@ -4176,6 +4195,7 @@ def create_subtask( } +@guard_task_write def set_task_epic( task_concept_id: str, epic_task_concept_id: str | None, @@ -4233,6 +4253,7 @@ def set_task_epic( return get_task(task_concept_id) +@guard_task_write def link_tasks( source_task_concept_id: str, target_task_concept_id: str, @@ -4305,6 +4326,7 @@ def link_tasks( } +@guard_task_write def unlink_tasks( source_task_concept_id: str, target_task_concept_id: str, @@ -4375,6 +4397,7 @@ def unlink_tasks( } +@guard_task_write def add_task_comment( task_concept_id: str, *, @@ -4486,6 +4509,7 @@ def find_task_comment_by_effect_fingerprint( return None +@guard_task_write def add_task_attachment_bytes( task_concept_id: str, *, @@ -4559,6 +4583,7 @@ def add_task_attachment_bytes( ) +@guard_task_write def add_task_attachment( task_concept_id: str, *, @@ -4690,6 +4715,7 @@ def list_task_attachments( } +@guard_task_write def add_task_worklog( task_concept_id: str, *, @@ -4742,6 +4768,7 @@ def add_task_worklog( return entry +@guard_task_write def record_task_history_event( task_concept_id: str, *, @@ -4825,6 +4852,7 @@ def get_task_history( } +@guard_task_write def update_task_fields( task_concept_id: str, *, @@ -5594,6 +5622,7 @@ def find_task_by_external_reference( return _build_task_response(doc) +@guard_task_write def upsert_task_external_reference( task_concept_id: str, *, @@ -5705,6 +5734,7 @@ def _task_doc_has_bulk_collection( ) +@guard_task_write def backfill_jira_migration_bulk_task_collections( *, dry_run: bool = True, @@ -5837,6 +5867,7 @@ def backfill_jira_migration_bulk_task_collections( } +@guard_task_write def bulk_update_tasks( task_concept_ids: Iterable[str], *, @@ -5890,6 +5921,7 @@ def bulk_update_tasks( } +@guard_task_write def delete_task(task_concept_id: str) -> bool: """Delete (archive) a task. @@ -5989,6 +6021,7 @@ def get_task_taxonomy() -> Dict[str, Any]: ] +@guard_task_write def refresh_coding_supervision_context( repair_task_id: str, *, diff --git a/src/backend/services/task_project_home_service.py b/src/backend/services/task_project_home_service.py new file mode 100644 index 00000000..386e33b3 --- /dev/null +++ b/src/backend/services/task_project_home_service.py @@ -0,0 +1,306 @@ +"""Explicit task-project write homes and a recoverable cutover barrier. + +Home routing is operational state, separate from Jira/Von tracking authority. +An interrupted admitted write retains its token: elapsed time cannot certify +that its effects stopped. Operator transitions require an empty operation set. +This service does not copy execution leases or authorise a task's execution. +""" + +from __future__ import annotations + +import os +import uuid +import inspect +from functools import wraps +from contextlib import contextmanager, ExitStack +from contextvars import ContextVar +from datetime import UTC, datetime +from typing import Any + +from pymongo import ReturnDocument +from pymongo.errors import DuplicateKeyError + +COLLECTION = "task_project_homes" +_ADMITTED: ContextVar[frozenset[str]] = ContextVar( + "task_project_home_admitted", default=frozenset() +) + + +class TaskProjectHomeError(ValueError): + def __init__(self, code: str, route: dict | None = None): + self.code = code + self.route = route or {} + super().__init__( + f"{code}: use task project home {self.route.get('home_url', 'recorded in its route')}" + ) + + +def _collection(): + from ..db.mongo_client import get_db + + db = get_db() + if db is None: + raise TaskProjectHomeError("task_home_unavailable") + return db[COLLECTION] + + +def local_node_id() -> str: + return os.environ.get("VON_TASK_HOME_NODE_ID", "").strip() + + +def get_project_home(project_id: str) -> dict | None: + row = _collection().find_one({"_id": project_id}) + if row is None: + return None + return {key: value for key, value in row.items() if key not in {"operations"}} + + +def guard_task_write(function): + """Share one admission across nested canonical task mutations.""" + signature = inspect.signature(function) + + @wraps(function) + def guarded(*args, **kwargs): + from .task_management_service import _get_task_doc + from ..security.access_control import can_access_concept + + arguments = signature.bind(*args, **kwargs).arguments + project_ids = {arguments.get("project_concept_id")} + fields = arguments.get("fields") or {} + if isinstance(fields, dict): + project_ids.add(fields.get("project_concept_id")) + task_ids = set() + for key, value in arguments.items(): + if key in { + "task_concept_id", + "source_task_concept_id", + "target_task_concept_id", + "parent_task_concept_id", + "epic_task_concept_id", + "repair_task_id", + "task_concept_ids", + "task_ids", + }: + task_ids.update( + [value] + if isinstance(value, str) + else value if isinstance(value, (list, tuple)) else [] + ) + if task_ids: + docs = [_get_task_doc(task_id)[1] for task_id in sorted(task_ids)] + project_ids.update( + (doc.get("metadata") or {}).get("project_concept_id") for doc in docs + ) + with ExitStack() as stack: + for project_id in sorted(project_ids - {None, ""}): + if not can_access_concept(project_id): + raise PermissionError("Task project is not accessible") + stack.enter_context(project_write(project_id)) + return function(*args, **kwargs) + + return guarded + + +def register_project_home( + *, + project_id: str, + node_id: str, + home_url: str, + source_identity: dict, + evidence: dict, +) -> dict: + """Operator-only initial registration; never overwrite a prior route.""" + if ( + not project_id + or not node_id + or not home_url + or not source_identity + or not evidence + ): + raise ValueError("Project, home, source identity and evidence are required") + row = { + "_id": project_id, + "schema_version": "task_project_home.v1", + "home_node_id": node_id, + "home_url": home_url, + "epoch": 1, + "state": "active", + "source_identity": source_identity, + "evidence": evidence, + "operations": {}, + "updated_at": datetime.now(UTC), + } + try: + _collection().insert_one(row) + except DuplicateKeyError: + existing = get_project_home(project_id) + if any(existing.get(k) != row[k] for k in ("home_node_id", "source_identity")): + raise TaskProjectHomeError("task_home_registration_conflict", existing) + return get_project_home(project_id) + + +@contextmanager +def project_write(project_id: str | None): + """Admit a canonical mutation at the local home, before any task effects.""" + if not project_id or project_id in _ADMITTED.get(): + yield + return + collection = _collection() + home = collection.find_one({"_id": project_id}) + if home is None: + # Existing projects remain compatible until an explicit registration. + yield + return + if home.get("state") != "active" or home.get("home_node_id") != local_node_id(): + raise TaskProjectHomeError("task_project_read_only_replica", home) + operation_id = uuid.uuid4().hex + admitted = collection.find_one_and_update( + { + "_id": project_id, + "epoch": home["epoch"], + "state": "active", + "home_node_id": local_node_id(), + }, + { + "$set": { + f"operations.{operation_id}": { + "started_at": datetime.now(UTC), + "pid": os.getpid(), + } + } + }, + return_document=ReturnDocument.AFTER, + ) + if admitted is None: + raise TaskProjectHomeError( + "task_home_changed_before_write", get_project_home(project_id) + ) + token = _ADMITTED.set(_ADMITTED.get() | {project_id}) + completed = False + try: + yield + completed = True + finally: + _ADMITTED.reset(token) + # A killed process or an uncertain failed write retains its token. + # Reconcile its effects before certifying the project quiescent. + if completed: + collection.update_one( + {"_id": project_id, "epoch": home["epoch"]}, + {"$unset": {f"operations.{operation_id}": ""}}, + ) + + +def reconcile_project_operation( + *, project_id: str, operation_id: str, expected_operation: dict, evidence: dict +) -> bool: + """Operator recovery after independently checking process and effect state.""" + if not evidence.get("process_drained") or not evidence.get("effects_reconciled"): + raise ValueError("Drain and canonical effect reconciliation are required") + if not operation_id.isalnum() or not expected_operation: + raise ValueError("Exact retained operation required") + result = _collection().update_one( + {"_id": project_id, f"operations.{operation_id}": expected_operation}, + { + "$unset": {f"operations.{operation_id}": ""}, + "$push": { + "operation_reconciliations": { + "operation_id": operation_id, + "evidence": evidence, + "at": datetime.now(UTC), + } + }, + }, + ) + return bool(result.matched_count) + + +def freeze_project_home( + *, project_id: str, expected_epoch: int, evidence: dict +) -> dict: + if not evidence: + raise ValueError("Freeze evidence is required") + row = _collection().find_one_and_update( + { + "_id": project_id, + "epoch": expected_epoch, + "state": "active", + "home_node_id": local_node_id(), + "operations": {}, + }, + { + "$set": { + "state": "frozen", + "freeze_evidence": evidence, + "updated_at": datetime.now(UTC), + } + }, + return_document=ReturnDocument.AFTER, + ) + if row is None: + current = get_project_home(project_id) + if ( + current + and current.get("state") == "frozen" + and current.get("epoch") == expected_epoch + ): + return current + raise TaskProjectHomeError("task_home_not_quiescent_or_changed", current) + return get_project_home(project_id) + + +def transfer_frozen_home( + *, + project_id: str, + expected_epoch: int, + node_id: str, + home_url: str, + snapshot_digest: str, + receipt: dict, +) -> dict: + """Fence this prior home after a verified destination receipt. + + Recovery never restores older task bytes. Returning the home requires a new + verified snapshot containing all changes accepted by its intervening home. + """ + if ( + not node_id + or not home_url + or len(snapshot_digest) != 64 + or not receipt.get("canonical_readback") + ): + raise ValueError("A verified destination snapshot receipt is required") + row = _collection().find_one_and_update( + { + "_id": project_id, + "epoch": expected_epoch, + "state": "frozen", + "operations": {}, + }, + { + "$set": { + "home_node_id": node_id, + "home_url": home_url, + "state": "replica", + "snapshot_digest": snapshot_digest, + "transfer_receipt": receipt, + "updated_at": datetime.now(UTC), + }, + "$inc": {"epoch": 1}, + }, + return_document=ReturnDocument.AFTER, + ) + if row is None: + current = get_project_home(project_id) + if current and all( + current.get(k) == v + for k, v in { + "home_node_id": node_id, + "state": "replica", + "snapshot_digest": snapshot_digest, + "epoch": expected_epoch + 1, + }.items() + ): + return current + raise TaskProjectHomeError("task_home_transfer_conflict", current) + return get_project_home(project_id) diff --git a/src/backend/services/task_project_transfer_service.py b/src/backend/services/task_project_transfer_service.py new file mode 100644 index 00000000..82cbc4df --- /dev/null +++ b/src/backend/services/task_project_transfer_service.py @@ -0,0 +1,272 @@ +"""Bounded, actor-scoped task-project snapshots for an explicit home transfer. + +Snapshots are data, never execution requests. Their file manifests refer to +separately verified bytes. Publication must be one database transaction; a +standalone MongoDB server cannot safely activate this adapter. +""" + +from __future__ import annotations + +import copy +from datetime import UTC, datetime +from typing import Any + +from bson import json_util + +from ..db.repositories.concepts_repository import ConceptsRepository +from ..db.repositories.text_value_repository import ( + TextRelationsRepository, + TextValuesRepository, +) +from ..security.access_control import can_access_concept, get_effective_user_concept_id +from ..security.visibility_predicates import ( + get_specific_to_org_values, + get_specific_to_user_values, +) +from .knowledge_federation.protocol import digest, encode, key, peer_config, sign +from .task_project_home_service import get_project_home + +VERSION = "task_project_snapshot.v1" +MAX_CONCEPTS = 20000 +MAX_BYTES = 256 * 1024 * 1024 +# Dispatch, supervision and execution state must be established at the new home. +EXECUTION_METADATA = { + "coding_supervision", + "coding_dispatch", + "dispatch_authority", + "task_execution", + "execution_lease", + "execution_state", +} +CONCEPT_FIELDS = { + "concept_id", + "guid", + "created_at", + "updated_at", + "relationships", + "metadata", + "attributes", + "system_tags", + "user_tags", +} + + +def portable(value: Any) -> Any: + """Preserve BSON dates and identifiers without local storage identities.""" + import json + + return json.loads(json_util.dumps(value)) + + +def decoded(value: Any) -> Any: + import json + + return json_util.loads(json.dumps(value)) + + +def _scope(document: dict) -> dict: + relationships = document.get("relationships") or {} + return { + "users": sorted(get_specific_to_user_values(relationships)), + "organisations": sorted(get_specific_to_org_values(relationships)), + } + + +def _semantic_concept(document: dict) -> dict: + result = copy.deepcopy({k: v for k, v in document.items() if k in CONCEPT_FIELDS}) + metadata = result.get("metadata", {}) + for field in EXECUTION_METADATA: + metadata.pop(field, None) + external = metadata.get("external_references", {}) + for field in EXECUTION_METADATA: + external.pop(field, None) + # Operational federation/dispatch receipts are local; preserve their source + # digest in the signed manifest rather than accepting them as authority. + metadata.pop("federation", None) + return result + + +def referenced_file_ids(value: Any) -> set[str]: + ids = set() + if isinstance(value, dict): + for field, item in value.items(): + if ( + field in {"file_copy_concept_id", "file_concept_id"} + and isinstance(item, str) + and item.startswith("#V#") + ): + ids.add(item) + else: + ids.update(referenced_file_ids(item)) + elif isinstance(value, list): + for item in value: + ids.update(referenced_file_ids(item)) + return ids + + +def capture_project(project_id: str, *, config: dict, recipient: str) -> dict: + """Read one admitted project through the current authenticated actor. + + The raw repository read follows exact actor admission so historical + relationships to absent participants are retained as provenance. It does + not import their accounts, memberships, prompts or execution state. + """ + actor = get_effective_user_concept_id() + settings = peer_config(config, "exports", recipient) + if not actor or project_id not in settings.get("task_projects", []): + raise PermissionError("Task project export is not admitted") + if not can_access_concept(project_id): + raise PermissionError("Task project is inaccessible") + collection = ConceptsRepository.collection() + project = collection.find_one({"concept_id": project_id}) + record = ((project or {}).get("attributes") or {}).get("task_project") + if not record or record.get("writer") != "von": + raise ValueError("A native-writer task project is required") + collection_ids = record.get("collection_concept_ids", []) + query = { + "$or": [ + {"concept_id": {"$in": [project_id, *collection_ids]}}, + {"metadata.project_concept_id": project_id}, + ] + } + documents = list(collection.find(query).limit(MAX_CONCEPTS + 1)) + if len(documents) > MAX_CONCEPTS: + raise ValueError("Task project exceeds the bounded concept inventory") + ids = {d["concept_id"] for d in documents} + if not set(collection_ids).issubset(ids): + raise ValueError("Project collection inventory is incomplete") + audiences = set(settings.get("audiences", [])) + for document in documents: + if not can_access_concept(document["concept_id"]): + raise PermissionError("Project contains an inaccessible record") + scope = _scope(document) + keys = {f"user:{v}" for v in scope["users"]} | { + f"org:{v}" for v in scope["organisations"] + } + if not keys or not keys.intersection(audiences): + raise PermissionError("Task record has no admitted private audience") + semantic = [_semantic_concept(d) for d in documents] + files = [] + for file_id in sorted(referenced_file_ids(semantic)): + if not can_access_concept(file_id): + raise PermissionError("A task source file is inaccessible") + document = collection.find_one({"concept_id": file_id}) + if not document: + raise ValueError("Missing task source file") + files.append(_semantic_concept(document)) + relations = list( + TextRelationsRepository.collection().find( + { + "subject_concept_id": { + "$in": sorted(ids | {d["concept_id"] for d in files}) + } + } + ) + ) + text_ids = {r["object_text_id"] for r in relations} + # Historic stores may use a string representation of a BSON ObjectId. + from bson import ObjectId + + query_ids = set(text_ids) + for value in text_ids: + if ObjectId.is_valid(str(value)): + query_ids.add(ObjectId(str(value))) + texts = list( + TextValuesRepository.collection().find({"_id": {"$in": list(query_ids)}}) + ) + if {str(v["_id"]) for v in texts} != {str(v) for v in text_ids}: + raise ValueError("Dangling task text reference in source snapshot") + body = portable( + { + "concepts": sorted(semantic, key=lambda d: d["concept_id"]), + "text_relations": sorted(relations, key=lambda r: str(r["_id"])), + "text_values": sorted(texts, key=lambda r: str(r["_id"])), + "files": files, + } + ) + if len(encode(body)) > MAX_BYTES: + raise ValueError("Task project exceeds the bounded snapshot size") + home = get_project_home(project_id) + payload = { + "schema_version": VERSION, + "origin": config["node_id"], + "recipient": recipient, + "project_id": project_id, + "actor": actor, + "captured_at": datetime.now(UTC).isoformat(), + "source_home": portable(home), + "digest": digest(body), + "body": body, + } + return sign(payload, key(settings)) + + +def verify_snapshot(envelope: dict, *, config: dict, origin: str) -> dict: + """Authenticate an admitted immutable snapshot, before any staging effects.""" + import hashlib + import hmac + + settings = peer_config(config, "imports", origin) + payload = envelope["payload"] + expected = hmac.new(key(settings), encode(payload), hashlib.sha256).hexdigest() + if not hmac.compare_digest(expected, envelope.get("signature", "")): + raise PermissionError("Invalid task snapshot signature") + if ( + payload.get("schema_version") != VERSION + or payload.get("origin") != origin + or payload.get("recipient") != config.get("node_id") + ): + raise PermissionError("Incompatible or misdirected task snapshot") + if payload.get("project_id") not in settings.get("task_projects", []): + raise PermissionError("Task project import is not admitted") + body = payload["body"] + if len(encode(body)) > MAX_BYTES or digest(body) != payload.get("digest"): + raise ValueError("Invalid task snapshot bytes or digest") + if set(body) != {"concepts", "files", "text_values", "text_relations"}: + raise ValueError("Unsupported task snapshot section") + records = body["concepts"] + project_id = payload["project_id"] + projects = [r for r in records if r["concept_id"] == project_id] + if len(projects) != 1: + raise ValueError("Missing task project record") + project = projects[0].get("attributes", {}).get("task_project", {}) + if project.get("writer") != "von": + raise ValueError("Task project writer is not Von") + collection_ids = set(project.get("collection_concept_ids", [])) + for record in records: + cid = record["concept_id"] + if cid == project_id: + continue + if cid in collection_ids: + if not record.get("attributes", {}).get("task_collection"): + raise ValueError("Invalid task collection record") + elif record.get("metadata", {}).get("project_concept_id") != project_id: + raise ValueError("Unrelated concept in task project snapshot") + subject_ids = {r["concept_id"] for r in [*records, *body["files"]]} + if any(r["subject_concept_id"] not in subject_ids for r in body["text_relations"]): + raise ValueError("Unrelated task text relation") + if {r["concept_id"] for r in body["files"]} != referenced_file_ids(records): + raise ValueError("Task file inventory differs from its references") + if len(records) > MAX_CONCEPTS or len({r["concept_id"] for r in records}) != len( + records + ): + raise ValueError("Invalid task snapshot concept inventory") + # Identity mappings are checked explicitly, independently of subscriptions. + # A source username/organisation slug cannot create local account authority. + mappings = settings.get("identity_map", {}) + for record in [*records, *body["files"]]: + scope = _scope(record) + audience_keys = {f"user:{v}" for v in scope["users"]} | { + f"org:{v}" for v in scope["organisations"] + } + if not audience_keys.intersection(settings.get("audiences", [])): + raise PermissionError("Task snapshot audience is outside the subscription") + for value in scope["users"] + scope["organisations"]: + if mappings.get(value) != value: + raise PermissionError("Unverified task audience identity mapping") + semantic = portable(_semantic_concept(decoded(record))) + if semantic != record: + raise ValueError( + "Snapshot contains execution or unsupported concept fields" + ) + return payload diff --git a/tests/backend/test_task_project_home_service.py b/tests/backend/test_task_project_home_service.py new file mode 100644 index 00000000..ce774ebb --- /dev/null +++ b/tests/backend/test_task_project_home_service.py @@ -0,0 +1,110 @@ +import mongomock +import pytest + +from src.backend.services import task_project_home_service as homes + + +@pytest.fixture +def store(monkeypatch): + collection = mongomock.MongoClient().test[homes.COLLECTION] + monkeypatch.setattr(homes, "_collection", lambda: collection) + monkeypatch.setenv("VON_TASK_HOME_NODE_ID", "atlas-mac") + homes.register_project_home( + project_id="#V#project", + node_id="atlas-mac", + home_url="https://source.test", + source_identity={"site": "jira.test", "id": "1"}, + evidence={"inventory_verified": True}, + ) + return collection + + +def test_freeze_drains_admitted_writes_and_blocks_later_writes(store): + with homes.project_write("#V#project"): + with homes.project_write("#V#project"): + assert len(store.find_one({})["operations"]) == 1 + with pytest.raises(homes.TaskProjectHomeError, match="not_quiescent"): + homes.freeze_project_home( + project_id="#V#project", expected_epoch=1, evidence={"checked": True} + ) + assert store.find_one({})["operations"] == {} + frozen = homes.freeze_project_home( + project_id="#V#project", expected_epoch=1, evidence={"checked": True} + ) + assert frozen["state"] == "frozen" + with pytest.raises(homes.TaskProjectHomeError, match="read_only_replica"): + with homes.project_write("#V#project"): + pytest.fail("The frozen source accepted a write") + + +def test_interrupted_write_requires_exact_effect_reconciliation(store): + with pytest.raises(RuntimeError): + with homes.project_write("#V#project"): + raise RuntimeError("lost database acknowledgement") + operations = store.find_one({})["operations"] + operation_id, operation = next(iter(operations.items())) + with pytest.raises(homes.TaskProjectHomeError): + homes.freeze_project_home( + project_id="#V#project", expected_epoch=1, evidence={"checked": True} + ) + with pytest.raises(ValueError): + homes.reconcile_project_operation( + project_id="#V#project", + operation_id=operation_id, + expected_operation=operation, + evidence={"process_drained": True}, + ) + assert homes.reconcile_project_operation( + project_id="#V#project", + operation_id=operation_id, + expected_operation=operation, + evidence={ + "process_drained": True, + "effects_reconciled": True, + "readback": "receipt", + }, + ) + homes.freeze_project_home( + project_id="#V#project", expected_epoch=1, evidence={"checked": True} + ) + + +def test_transfer_is_fenced_idempotent_and_rejects_stale_rollback(store, monkeypatch): + homes.freeze_project_home( + project_id="#V#project", expected_epoch=1, evidence={"checked": True} + ) + args = dict( + project_id="#V#project", + expected_epoch=1, + node_id="dgx", + home_url="https://destination.test", + snapshot_digest="a" * 64, + receipt={"canonical_readback": True, "snapshot_digest": "a" * 64}, + ) + moved = homes.transfer_frozen_home(**args) + assert homes.transfer_frozen_home(**args) == moved + assert moved["epoch"] == 2 + with pytest.raises(homes.TaskProjectHomeError): + with homes.project_write("#V#project"): + pytest.fail("Old home accepted an independent write") + with pytest.raises(homes.TaskProjectHomeError): + homes.transfer_frozen_home( + **{**args, "node_id": "atlas-mac", "snapshot_digest": "b" * 64} + ) + # Even identifying as the new node cannot activate an unverified replica. + monkeypatch.setenv("VON_TASK_HOME_NODE_ID", "dgx") + with pytest.raises(homes.TaskProjectHomeError): + with homes.project_write("#V#project"): + pytest.fail("A replica became authoritative by environment selection") + + +def test_source_identity_collision_is_not_an_overwrite(store): + with pytest.raises(homes.TaskProjectHomeError, match="registration_conflict"): + homes.register_project_home( + project_id="#V#project", + node_id="atlas-mac", + home_url="https://source.test", + source_identity={"site": "another.test", "id": "1"}, + evidence={"checked": True}, + ) + assert store.find_one({})["source_identity"]["site"] == "jira.test" diff --git a/tests/backend/test_task_project_transfer_service.py b/tests/backend/test_task_project_transfer_service.py new file mode 100644 index 00000000..1d2b6111 --- /dev/null +++ b/tests/backend/test_task_project_transfer_service.py @@ -0,0 +1,156 @@ +import copy +from datetime import UTC, datetime + +import mongomock +import pytest + +from src.backend.services import task_project_transfer_service as transfer +from src.backend.services.knowledge_federation.protocol import digest, sign + +USER = "#V#owner" +PROJECT = "#V#project" +TASK = "#V#task" + + +@pytest.fixture +def fixture(monkeypatch, tmp_path): + database = mongomock.MongoClient().test + monkeypatch.setattr( + transfer.ConceptsRepository, "collection", lambda: database.concepts + ) + monkeypatch.setattr( + transfer.TextRelationsRepository, "collection", lambda: database.text_relations + ) + monkeypatch.setattr( + transfer.TextValuesRepository, "collection", lambda: database.text_values + ) + monkeypatch.setattr(transfer, "can_access_concept", lambda concept_id: True) + monkeypatch.setattr(transfer, "get_effective_user_concept_id", lambda: USER) + monkeypatch.setattr(transfer, "get_project_home", lambda project_id: None) + secret = tmp_path / "key" + secret.write_text("01" * 32) + secret.chmod(0o600) + settings = { + "enabled": True, + "key_file": str(secret), + "task_projects": [PROJECT], + "audiences": ["user:" + USER], + "identity_map": {USER: USER}, + } + source = {"node_id": "atlas", "exports": {"dgx": settings}} + destination = {"node_id": "dgx", "imports": {"atlas": settings}} + audience = {"#V#specific_to_user": [USER]} + database.concepts.insert_many( + [ + { + "concept_id": PROJECT, + "relationships": audience, + "attributes": { + "task_project": {"writer": "von", "collection_concept_ids": []} + }, + }, + { + "concept_id": TASK, + "relationships": { + **audience, + "#V#has_created_by": ["#V#historical_author"], + }, + "metadata": { + "project_concept_id": PROJECT, + "comments": [ + {"author": "#V#historical_author", "body": "Retain me"} + ], + "coding_dispatch": {"authority": "must not transfer"}, + "external_references": { + "jira": {"issue_id": "42"}, + "coding_supervision": {"state": "running"}, + }, + }, + "created_at": datetime(2024, 1, 1, tzinfo=UTC), + }, + ] + ) + return database, source, destination + + +def test_signed_snapshot_preserves_authors_and_dates_without_execution_state(fixture): + database, source, destination = fixture + envelope = transfer.capture_project(PROJECT, config=source, recipient="dgx") + payload = transfer.verify_snapshot(envelope, config=destination, origin="atlas") + task = next( + r + for r in transfer.decoded(payload["body"])["concepts"] + if r["concept_id"] == TASK + ) + assert task["relationships"]["#V#has_created_by"] == ["#V#historical_author"] + assert task["metadata"]["comments"][0]["author"] == "#V#historical_author" + assert task["created_at"].year == 2024 + assert "coding_dispatch" not in task["metadata"] + assert set(task["metadata"]["external_references"]) == {"jira"} + assert database.concepts.find_one({"concept_id": TASK})["metadata"][ + "coding_dispatch" + ] + assert ( + transfer.capture_project(PROJECT, config=source, recipient="dgx")["payload"][ + "digest" + ] + == payload["digest"] + ) + database.concepts.update_one( + {"concept_id": TASK}, {"$push": {"metadata.comments": {"body": "Later change"}}} + ) + assert ( + transfer.capture_project(PROJECT, config=source, recipient="dgx")["payload"][ + "digest" + ] + != payload["digest"] + ) + + +def test_tampering_and_wrong_recipient_are_rejected(fixture): + _, source, destination = fixture + envelope = transfer.capture_project(PROJECT, config=source, recipient="dgx") + envelope["payload"]["project_id"] = "#V#another" + with pytest.raises(PermissionError, match="signature"): + transfer.verify_snapshot(envelope, config=destination, origin="atlas") + envelope = transfer.capture_project(PROJECT, config=source, recipient="dgx") + destination["node_id"] = "another" + with pytest.raises(PermissionError, match="misdirected"): + transfer.verify_snapshot(envelope, config=destination, origin="atlas") + + +def test_same_slug_is_not_an_audience_identity_proof(fixture): + _, source, destination = fixture + envelope = transfer.capture_project(PROJECT, config=source, recipient="dgx") + destination["imports"]["atlas"]["identity_map"] = {} + with pytest.raises(PermissionError, match="identity mapping"): + transfer.verify_snapshot(envelope, config=destination, origin="atlas") + + +def test_inaccessible_member_prevents_partial_export(fixture, monkeypatch): + _, source, _ = fixture + monkeypatch.setattr(transfer, "can_access_concept", lambda cid: cid != TASK) + with pytest.raises(PermissionError, match="inaccessible record"): + transfer.capture_project(PROJECT, config=source, recipient="dgx") + + +def test_admitted_peer_cannot_include_an_unrelated_concept(fixture): + _, source, destination = fixture + payload = transfer.capture_project(PROJECT, config=source, recipient="dgx")[ + "payload" + ] + payload["body"]["concepts"].append( + {"concept_id": "#V#production_prompt", "attributes": {}} + ) + payload["digest"] = digest(payload["body"]) + envelope = sign(payload, bytes.fromhex("01" * 32)) + with pytest.raises(ValueError, match="Unrelated concept"): + transfer.verify_snapshot(envelope, config=destination, origin="atlas") + + +def test_subscription_is_checked_independently_of_identity_map(fixture): + _, source, destination = fixture + envelope = transfer.capture_project(PROJECT, config=source, recipient="dgx") + destination["imports"]["atlas"]["audiences"] = ["user:#V#someone_else"] + with pytest.raises(PermissionError, match="outside the subscription"): + transfer.verify_snapshot(envelope, config=destination, origin="atlas") From 53bae5dcea532cf563d170add1d2e8a565f40c61 Mon Sep 17 00:00:00 2001 From: witbrock Date: Wed, 16 Sep 2026 12:02:16 -0400 Subject: [PATCH 2/3] Publish task-project snapshots atomically and require current dispatch authority --- docs/design_index.md | 1 + .../engineering/task_project_home_transfer.md | 95 ++++ scripts/codex_von_worker.py | 48 +- scripts/durable_claim_admission.py | 17 +- scripts/restore_von_db.py | 24 +- scripts/transfer_task_project.py | 196 ++++++++ .../db/repositories/concepts_repository.py | 35 +- .../db/repositories/text_value_repository.py | 32 +- .../services/jira_source_retention_service.py | 9 +- .../task_dispatch_authority_service.py | 72 +++ .../services/task_management_service.py | 57 ++- .../task_project_file_transfer_service.py | 55 +++ .../services/task_project_handoff_service.py | 170 +++++++ .../services/task_project_home_service.py | 172 ++++++- .../task_project_publication_service.py | 442 ++++++++++++++++++ src/backend/services/task_project_service.py | 40 +- .../services/task_project_transfer_service.py | 49 +- .../workflows/durable/instance_manager.py | 19 +- .../workflows/durable/task_ownership.py | 4 +- .../workflows/durable/worker_identity.py | 1 + .../static/js/components/taskPanel.js | 25 +- tests/backend/test_codex_von_worker.py | 13 +- ...est_concepts_repository_embedding_queue.py | 6 + .../test_jira_source_retention_service.py | 32 ++ .../test_task_dispatch_authority_service.py | 76 +++ tests/backend/test_task_management_service.py | 5 + .../test_task_project_handoff_service.py | 125 +++++ .../backend/test_task_project_home_service.py | 85 ++++ ...st_task_project_publication_transaction.py | 226 +++++++++ tests/backend/test_task_project_service.py | 34 ++ .../test_task_project_transfer_service.py | 5 +- tests/test_restore_von_db.py | 29 +- 32 files changed, 2126 insertions(+), 73 deletions(-) create mode 100644 docs/engineering/task_project_home_transfer.md create mode 100644 scripts/transfer_task_project.py create mode 100644 src/backend/services/task_dispatch_authority_service.py create mode 100644 src/backend/services/task_project_file_transfer_service.py create mode 100644 src/backend/services/task_project_handoff_service.py create mode 100644 src/backend/services/task_project_publication_service.py create mode 100644 tests/backend/test_task_dispatch_authority_service.py create mode 100644 tests/backend/test_task_project_handoff_service.py create mode 100644 tests/backend/test_task_project_publication_transaction.py diff --git a/docs/design_index.md b/docs/design_index.md index df87eb0e..27afaafc 100644 --- a/docs/design_index.md +++ b/docs/design_index.md @@ -127,6 +127,7 @@ from `AGENTS.md`. | [Agent memory and enduring knowledge](engineering/agent_memory_and_enduring_knowledge.md) | Active design guide | Normative only within its stated memory/retrieval scope | | [Role-learning convergence](engineering/role_learning_convergence.md) | Active programme design and guide | Canonical programme direction under `AGENTS.md` for claims about enduring role competence; no runtime activation, universal per-PR gate or demonstrated role capability is implied | | [Knowledge federation pilot](engineering/knowledge_federation_pilot.md) | Bounded implementation and authority contract | JVNAUTOSCI-2730/2731: automatic scoped discovery, conversation recall and on-demand private artefact copies; opt-in complete-slice exchange between independent stores; source audiences, atomic receipt/read projection, expiry, recovery and later SC448086 admission. Not general database replication or execution federation. | +| [Task-project home transfer](engineering/task_project_home_transfer.md) | Bounded task-project operations and recovery | Signed project snapshots, preserved audiences and archives, explicit Atlas-to-DGX write-home handoff, current-actor dispatch and interruption recovery. | | [Contextual knowledge evolution](engineering/contextual_knowledge_evolution.md) | Active design guide | Advisory under `AGENTS.md` when assertion/context distinctions are material; it does not select a microtheory formalism or prove present implementation | | [Assertion and propositional-sentence ontology](engineering/assertion_and_propositional_sentence_ontology.md) | Active design and partial implementation boundary | Canonical model and phased integration plan; the minimal public vocabulary is live, while assertion-record typing, context semantics, general logical storage, and reasoning remain future work | | [External conversation import](engineering/external_conversation_import.md) | Active design and implementation boundary | Canonical reference for source-neutral transcript packages, actor-scoped read-only projections, provider adapters, append-only resynchronisation, durable machine-wide batches, and native continuation; verify supported provider export shapes against current fixtures and live evidence | diff --git a/docs/engineering/task_project_home_transfer.md b/docs/engineering/task_project_home_transfer.md new file mode 100644 index 00000000..be1d89dc --- /dev/null +++ b/docs/engineering/task_project_home_transfer.md @@ -0,0 +1,95 @@ +# Task-project home transfer + +This bounded operational slice of the federation programme (JVNAUTOSCI-2615) +keeps MongoDB and transfers selected native-writer projects. It does not replace +the broader federation plan or implement general database replication. + +## Contract and authority + +`writer=von` chooses the tracking application. `task_project_homes` separately +records the physical write/execution home, epoch, route and state. Private peer +configuration admits exact projects and audiences; verified identity mappings +are independent of subscriptions. Task IDs, original authors, Jira aliases, +comments/history/relationships, project membership and source file bytes survive. +Accounts, memberships, login bindings, dispatch receipts and execution state do +not travel with a snapshot. Current canonical assignment creates local dispatch +authority; a copied assignment cannot start a coding agent. + +Canonical task, concept and text-relation mutations obey the project barrier. +The durable worker checks the home before claiming and again before actions. +The claim-admission validator can require `task_project_home_v1` to exclude older +workers. This does not constrain arbitrary administrator database writes. +Before first registration, identify and drain all previous writers and in-flight +claims; update active consumers and fence legacy worker claims. An in-flight +external action is not cancelled merely by changing a home record. + +## Transfer order + +Use `scripts/transfer_task_project.py` under the explicitly selected environment +with a private federation config, `VON_TASK_HOME_NODE_ID`, actor and original +organisation scope. The script is an operator CLI, not a public authority API. +Each output is owner-only and should be retained with the task's audit evidence. + +1. Inventory the exact projects, audiences, identities, collision histories, + active execution and source/destination recovery evidence. Verify old records + by Jira source identity and history retention, not concept slug or date alone. +2. Deploy home-aware canonical services and consumers. Require the worker + capability at the shared claim boundary after capable workers are available. +3. `register` the source home with the evidence file. `freeze` the exact epoch + only after writers/claims have drained and every retained operation has been + reconciled. Failed writes retain operation tokens; elapsed time is not proof + their effects stopped. +4. `capture` a fresh signed snapshot while frozen. An active/preview snapshot + cannot be published. Transfer the snapshot and private hash-named file cache + through an authenticated operator channel. Keep credentials on their hosts. +5. At the destination, `preflight` verifies identities/history and records exact + current concept and text digests. `stage-files` verifies all file hashes and + sizes before and after destination storage. A missing/corrupt file stops the + transfer without publishing concepts. +6. `publish` rechecks the preflight inside one MongoDB transaction, retaining the + prior complete view if interrupted. Transactions require a replica set; a + standalone server is rejected. The complete snapshot and receipt become + visible together with a **prepared**, non-writable, non-executing home. +7. Canonically read back project/task text, relationships, membership, aliases, + audiences and file bytes. Obtain the signed destination `receipt`. +8. At the source, `release` compares a fresh frozen capture with that receipt, + moves the route to a read-only replica and returns a signed release. +9. At the destination, `activate` requires the exact signed source release, + publication receipt, digest and epoch. Only now can canonical writes and + deliberate assignment proceed. Verify public search, native create/edit and + an explicitly assigned harmless coding-agent canary through actual result. +10. Repeat for the remaining admitted projects in their own scopes. Report + per-project completion and outstanding exceptions, not only pilot success. + +## Interruption and recovery + +Before publication, retry capture/staging after inspecting the retained state. +An interrupted transaction leaves the previous destination view intact. A +committed publication can be replayed idempotently; its receipt is returned +without overwriting later edits. Preflight rejects concurrent destination text +or concept changes and requires a new reconciliation. + +Between publication and release, both homes remain non-writable. Retry the +exact receipt after diagnosing the failure. Between release and activation, +the source remains a read-only replica; retry the exact signed release. There +is no timeout-based ownership takeover. Stale epochs cannot reactivate a prior +home. After activation, recover the current destination in place, retaining +all subsequent writes. Returning ownership is a new explicitly reconciled +transfer of the latest state; restoring the old source snapshot is prohibited. +The current CLI does not automate a reverse transfer into an existing home. + +Mongo recovery must be tested in a separate authenticated database. A live dump +is not a whole-database point-in-time snapshot. Preserve the existing data +volume and take a cold copy before changing a standalone server's topology. +A topology rollback must retain the latest data volume, not replace it with a +pre-transfer backup after new work has been accepted. Keep blob bytes with the +paired database recovery evidence. + +## Validation boundary + +Targeted tests cover audience/signature denial, retained uncertain operations, +write/claim fencing, actor-bound dispatch, signed handoff, stale-epoch rejection, +and real-Mongo transaction interruption/retry with later-write preservation. +Real project rehearsal and public/canonical read-back are required for a live +transfer claim. Fixture home certificates are labelled fixture-only and never +establish that a production source was released. diff --git a/scripts/codex_von_worker.py b/scripts/codex_von_worker.py index efcd134d..dee3db47 100755 --- a/scripts/codex_von_worker.py +++ b/scripts/codex_von_worker.py @@ -187,15 +187,29 @@ def require_task_execution_preferences(task): def authorised_task(task, config): - return ( - task.get("assignee_concept_id") == config["agent_id"] + if ( + task.get("assignee_concept_id") != config["agent_id"] + or task.get("organisation_concept_id") not in (None, config["organisation_id"]) + ): + return False + if not config.get("supervision_enabled") and task.get( + "report_to_concept_id" + ) not in (None, config["delegator_id"]): + return False + from src.backend.services.task_dispatch_authority_service import ( + has_assignment_record, + is_authorised_assignment, + ) + if ( + not task.get("federation") and task.get("organisation_concept_id") == config["organisation_id"] and task.get("created_by_concept_id") == config["delegator_id"] - and ( - config.get("supervision_enabled") - or task.get("report_to_concept_id") in (None, config["delegator_id"]) - ) - ) + and not has_assignment_record(task["task_concept_id"]) + ): + return True + if not task.get("dispatch_requested"): + return False + return is_authorised_assignment(task, config["delegator_id"]) def referenced_tasks(config, api, supplied, task_ids=()): @@ -435,8 +449,8 @@ def check_task(config, api, task_id): while True: page = api.tasks.search_tasks( assignee_concept_id=config["agent_id"], - created_by_concept_id=config["delegator_id"], organisation_concept_id=config["organisation_id"], + organisation_scope_mode="current_plus_unscoped", statuses=[task["status"]], limit=200, offset=offset, @@ -504,10 +518,18 @@ def task(self, task_id): def native_writer(self, task): project_id = task.get("project_concept_id") if project_id: - from src.backend.services.task_project_service import get_task_project - - project = get_task_project(project_id) - return bool(project and project.get("writer") == "von") + from src.backend.services.task_project_service import task_execution_home + + project = task_execution_home(task["task_concept_id"]) + home = project.get("write_home") if project else None + return bool( + project + and project.get("writer") == "von" + and ( + home is None + or home.get("writable_here") is True + ) + ) return not task.get("is_imported_jira_task", False) def pending(self): @@ -515,8 +537,8 @@ def pending(self): while True: page = self.tasks.search_tasks( assignee_concept_id=self.config["agent_id"], - created_by_concept_id=self.config["delegator_id"], organisation_concept_id=self.config["organisation_id"], + organisation_scope_mode="current_plus_unscoped", statuses=sorted(ACTIVE), limit=200, offset=offset, diff --git a/scripts/durable_claim_admission.py b/scripts/durable_claim_admission.py index 970208f4..1d1af576 100644 --- a/scripts/durable_claim_admission.py +++ b/scripts/durable_claim_admission.py @@ -16,10 +16,17 @@ def install_admission( - db, *, collection="workflow_instances", worker_pattern="^worker_", apply=False + db, + *, + collection="workflow_instances", + worker_pattern="^worker_", + apply=False, + required_capability="durable_task_ownership_v1", ): options = db[collection].options() - rule = claim_admission_validator(worker_pattern) + rule = claim_admission_validator( + worker_pattern, required_capability=required_capability + ) existing = options.get("validator", {}) validator = ( existing @@ -55,10 +62,14 @@ def install_admission( parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--apply", action="store_true") parser.add_argument("--worker-pattern", default="^worker_") + parser.add_argument("--required-capability", default="durable_task_ownership_v1") parser.add_argument("--receipt", type=Path, required=True) args = parser.parse_args() result = install_admission( - get_db(), worker_pattern=args.worker_pattern, apply=args.apply + get_db(), + worker_pattern=args.worker_pattern, + apply=args.apply, + required_capability=args.required_capability, ) args.receipt.write_text(json.dumps(result, indent=2, default=str)) print(json.dumps(result, default=str)) diff --git a/scripts/restore_von_db.py b/scripts/restore_von_db.py index aa0b1252..a9682e9b 100644 --- a/scripts/restore_von_db.py +++ b/scripts/restore_von_db.py @@ -262,6 +262,28 @@ def _resolve_blob_store_for_restore(backend: str): os.environ["VON_BLOB_STORE_BACKEND"] = orig +def _uri_for_namespace_restore(uri: str) -> str: + """Let --nsFrom/--nsTo select databases while preserving authentication. + + A database path in the URI makes mongorestore treat --dir as a single + database directory and skip its database subdirectories with exit code 0. + """ + from urllib.parse import parse_qsl, unquote, urlencode, urlsplit, urlunsplit + + parts = urlsplit(uri) + database = unquote(parts.path.lstrip("/")) + query = parse_qsl(parts.query, keep_blank_values=True) + if ( + database + and parts.username + and not any(k.lower() == "authsource" for k, _ in query) + ): + query.append(("authSource", database)) + return urlunsplit( + (parts.scheme, parts.netloc, "/", urlencode(query), parts.fragment) + ) + + def _run_mongorestore( *, mongo_uri: str, @@ -302,7 +324,7 @@ def link_or_copy(source, destination): ) config_path = config_root / "config.yml" config_path.write_text( - f"uri: {json.dumps(mongo_uri)}\n", + f"uri: {json.dumps(_uri_for_namespace_restore(mongo_uri))}\n", encoding="utf-8", ) os.chmod(config_path, 0o600) diff --git a/scripts/transfer_task_project.py b/scripts/transfer_task_project.py new file mode 100644 index 00000000..b335b52f --- /dev/null +++ b/scripts/transfer_task_project.py @@ -0,0 +1,196 @@ +"""Operate a bounded task-project transfer using canonical federation services. + +Run in the selected source or destination environment. Config/evidence/snapshot +files are private operator inputs. Commands never enable a worker or invoke a +model. See docs/engineering/task_project_home_transfer.md for ordering/recovery. +""" + +from __future__ import annotations + +import argparse +import json +import os +import sys +from pathlib import Path + +ROOT = Path(__file__).resolve().parents[1] +if str(ROOT) not in sys.path: + sys.path.insert(0, str(ROOT)) + + +def main(): + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument( + "operation", + choices=[ + "register", + "freeze", + "capture", + "preflight", + "stage-files", + "publish", + "receipt", + "release", + "activate", + ], + ) + parser.add_argument("--config", required=True) + parser.add_argument("--peer", required=True) + parser.add_argument("--actor", required=True) + parser.add_argument("--organisation") + parser.add_argument("--project") + parser.add_argument("--input") + parser.add_argument("--preflight") + parser.add_argument("--staged-files") + parser.add_argument("--cache") + parser.add_argument("--blob-root") + parser.add_argument("--epoch", type=int) + parser.add_argument("--output", required=True) + args = parser.parse_args() + os.umask(0o077) + from src.backend.services.knowledge_federation.protocol import ( + load_config, + private_file, + peer_config, + ) + from src.backend.security.access_control import ( + override_current_actor, + force_access_control_enforcement, + ) + from src.backend.services.task_project_transfer_service import ( + capture_project, + verify_snapshot, + portable, + decoded, + ) + from src.backend.services import task_project_home_service as homes + from src.backend.services import task_project_publication_service as publication + from src.backend.services import task_project_handoff_service as handoff + from src.backend.services.task_project_file_transfer_service import ( + stage_project_files, + cached_source_bytes, + ) + from src.backend.services.blob_store import LocalBlobStore + from src.backend.db.mongo_client import get_db + from src.backend.services.task_project_service import get_task_project + + def read(path): + if not path: + raise ValueError("Required private input file is missing") + return json.loads(private_file(str(Path(path).resolve())).read_text()) + + config = load_config(args.config) + if homes.local_node_id() != config["node_id"]: + raise ValueError("VON_TASK_HOME_NODE_ID must identify the selected environment") + with ( + override_current_actor(args.actor, args.organisation), + force_access_control_enforcement(), + ): + operation = args.operation + if operation in {"register", "freeze"}: + settings = peer_config(config, "exports", args.peer) + if args.project not in settings.get("task_projects", []): + raise PermissionError("Project not admitted for transfer") + project = get_task_project(args.project) + evidence = read(args.input) + if operation == "register": + result = homes.register_project_home( + project_id=args.project, + node_id=config["node_id"], + home_url=config["home_url"], + source_identity={ + k: project[k] for k in ("source_site", "source_id") + }, + evidence=evidence, + ) + else: + result = homes.freeze_project_home( + project_id=args.project, + expected_epoch=args.epoch, + evidence=evidence, + ) + elif operation == "capture": + result = capture_project(args.project, config=config, recipient=args.peer) + elif operation in {"preflight", "stage-files", "publish"}: + envelope = read(args.input) + payload = verify_snapshot(envelope, config=config, origin=args.peer) + if payload["actor"] != args.actor: + raise PermissionError("Snapshot actor differs from operator scope") + if operation == "preflight": + body = decoded(payload["body"]) + documents = body["concepts"] + body["files"] + ids = [r["concept_id"] for r in documents] + database = get_db() + existing = { + r["concept_id"]: r + for r in database.concepts.find({"concept_id": {"$in": ids}}) + } + for document in documents: + if document["concept_id"] in existing: + publication.check_identity_and_retention( + existing[document["concept_id"]], document + ) + result = { + "snapshot_digest": payload["digest"], + "expected_destination_digests": { + cid: publication.concept_digest(existing.get(cid)) + for cid in ids + }, + "expected_destination_text_digest": publication.destination_text_digest( + database, ids + ), + } + else: + if not args.blob_root: + raise ValueError("Explicit destination blob root required") + store = LocalBlobStore(Path(args.blob_root)) + if operation == "stage-files": + if not args.cache: + raise ValueError( + "Explicit private source recovery cache required" + ) + result = stage_project_files( + envelope, + config=config, + origin=args.peer, + blob_store=store, + source_bytes=cached_source_bytes(Path(args.cache)), + ) + else: + preflight = read(args.preflight) + if preflight.pop("snapshot_digest") != payload["digest"]: + raise ValueError("Preflight belongs to another snapshot") + result = publication.publish_project_snapshot( + envelope, + config=config, + origin=args.peer, + staged_files=read(args.staged_files), + blob_store=store, + **preflight, + ) + elif operation == "receipt": + result = handoff.publication_receipt( + read(args.input)["_id"], config=config, origin=args.peer + ) + elif operation == "release": + result = handoff.release_source( + read(args.input), config=config, destination=args.peer + ) + else: + result = handoff.activate_destination( + read(args.input), config=config, origin=args.peer + ) + output = Path(args.output).resolve() + temporary = output.with_suffix(output.suffix + ".tmp") + temporary.write_text(json.dumps(portable(result), separators=(",", ":"))) + temporary.chmod(0o600) + temporary.replace(output) + print( + json.dumps( + {"operation": args.operation, "output": str(output), "completed": True} + ) + ) + + +if __name__ == "__main__": + main() diff --git a/src/backend/db/repositories/concepts_repository.py b/src/backend/db/repositories/concepts_repository.py index 267c72d8..6ac14a2b 100644 --- a/src/backend/db/repositories/concepts_repository.py +++ b/src/backend/db/repositories/concepts_repository.py @@ -249,7 +249,10 @@ def insert_one(document: Dict[str, Any]): raise RuntimeError("Concepts collection not available") prepared = dict(document) prepared.setdefault("embedding_status", _EMBEDDING_STATUS_PENDING) - return coll.insert_one(prepared) + from ...services.task_project_home_service import concept_mutation + + with concept_mutation(coll, {}, document=prepared): + return coll.insert_one(prepared) @staticmethod def update_one( @@ -261,7 +264,10 @@ def update_one( update = _sanitize_update_payload(update) update = _prepare_concept_embedding_status(update) query = apply_concept_query_filter(filter or {}) - return coll.update_one(query, update, upsert=upsert) + from ...services.task_project_home_service import concept_mutation + + with concept_mutation(coll, query, update=update, many=False) as fenced_query: + return coll.update_one(fenced_query, update, upsert=upsert) @staticmethod def update_many(filter: Dict[str, Any], update: Dict[str, Any]): @@ -271,7 +277,10 @@ def update_many(filter: Dict[str, Any], update: Dict[str, Any]): update = _sanitize_update_payload(update) update = _prepare_concept_embedding_status(update) query = apply_concept_query_filter(filter or {}) - return coll.update_many(query, update) + from ...services.task_project_home_service import concept_mutation + + with concept_mutation(coll, query, update=update, many=True) as fenced_query: + return coll.update_many(fenced_query, update) @staticmethod def find_one_and_update( @@ -283,9 +292,12 @@ def find_one_and_update( update = _sanitize_update_payload(update) update = _prepare_concept_embedding_status(update) query = apply_concept_query_filter(filter or {}) - result = coll.find_one_and_update( - query, update, return_document=return_document - ) + from ...services.task_project_home_service import concept_mutation + + with concept_mutation(coll, query, update=update) as fenced_query: + result = coll.find_one_and_update( + fenced_query, update, return_document=return_document + ) return sanitize_concept_document(result) @staticmethod @@ -294,7 +306,10 @@ def delete_one(filter: Dict[str, Any]): if coll is None: raise RuntimeError("Concepts collection not available") query = apply_concept_query_filter(filter or {}) - return coll.delete_one(query) + from ...services.task_project_home_service import concept_mutation + + with concept_mutation(coll, query, many=False) as fenced_query: + return coll.delete_one(fenced_query) @staticmethod def delete_many(filter: Dict[str, Any]): @@ -302,7 +317,10 @@ def delete_many(filter: Dict[str, Any]): if coll is None: raise RuntimeError("Concepts collection not available") query = apply_concept_query_filter(filter or {}) - return coll.delete_many(query) + from ...services.task_project_home_service import concept_mutation + + with concept_mutation(coll, query, many=True) as fenced_query: + return coll.delete_many(fenced_query) @staticmethod def count_documents(filter: Dict[str, Any]) -> int: @@ -459,6 +477,7 @@ def _collect_targets( from ...services.concept_predicate_metadata_service import ( get_relationship_kinds, ) + kinds: tuple[str, ...] = get_relationship_kinds() except Exception: kinds = RELATIONSHIP_KINDS diff --git a/src/backend/db/repositories/text_value_repository.py b/src/backend/db/repositories/text_value_repository.py index 34e3d50b..13970bde 100644 --- a/src/backend/db/repositories/text_value_repository.py +++ b/src/backend/db/repositories/text_value_repository.py @@ -208,7 +208,10 @@ def insert_one(document: Dict[str, Any]): coll = TextRelationsRepository.collection() if coll is None: raise RuntimeError("text_relations collection not available") - return coll.insert_one(document) + from ...services.task_project_home_service import text_relation_mutation + + with text_relation_mutation(coll, {}, document=document): + return coll.insert_one(document) @staticmethod def update_one( @@ -217,14 +220,20 @@ def update_one( coll = TextRelationsRepository.collection() if coll is None: raise RuntimeError("text_relations collection not available") - return coll.update_one(filter, update, upsert=upsert) + from ...services.task_project_home_service import text_relation_mutation + + with text_relation_mutation(coll, filter, update=update) as fenced_query: + return coll.update_one(fenced_query, update, upsert=upsert) @staticmethod def update_many(filter: Dict[str, Any], update: Dict[str, Any]): coll = TextRelationsRepository.collection() if coll is None: raise RuntimeError("text_relations collection not available") - return coll.update_many(filter, update) + from ...services.task_project_home_service import text_relation_mutation + + with text_relation_mutation(coll, filter, update=update) as fenced_query: + return coll.update_many(fenced_query, update) @staticmethod def find_one_and_update( @@ -233,21 +242,32 @@ def find_one_and_update( coll = TextRelationsRepository.collection() if coll is None: return None - return coll.find_one_and_update(filter, update, return_document=return_document) + from ...services.task_project_home_service import text_relation_mutation + + with text_relation_mutation(coll, filter, update=update) as fenced_query: + return coll.find_one_and_update( + fenced_query, update, return_document=return_document + ) @staticmethod def delete_one(filter: Dict[str, Any]): coll = TextRelationsRepository.collection() if coll is None: raise RuntimeError("text_relations collection not available") - return coll.delete_one(filter) + from ...services.task_project_home_service import text_relation_mutation + + with text_relation_mutation(coll, filter) as fenced_query: + return coll.delete_one(fenced_query) @staticmethod def delete_many(filter: Dict[str, Any]): coll = TextRelationsRepository.collection() if coll is None: raise RuntimeError("text_relations collection not available") - return coll.delete_many(filter) + from ...services.task_project_home_service import text_relation_mutation + + with text_relation_mutation(coll, filter) as fenced_query: + return coll.delete_many(fenced_query) @staticmethod def count_documents(filter: Dict[str, Any]) -> int: diff --git a/src/backend/services/jira_source_retention_service.py b/src/backend/services/jira_source_retention_service.py index 3d38dc67..75b16d93 100644 --- a/src/backend/services/jira_source_retention_service.py +++ b/src/backend/services/jira_source_retention_service.py @@ -671,6 +671,11 @@ def store_source_archive( raw = canonical_json(archive) digest = hashlib.sha256(raw).hexdigest() data = gzip.compress(raw, mtime=0) + # Python 3.11/3.12 may emit the platform OS byte with mtime=0; later + # versions emit 255. Keep the header portable and address stored bytes + # independently of raw JSON so another compressor cannot replace them. + data = data[:9] + b"\xff" + data[10:] + blob_digest = hashlib.sha256(data).hexdigest() source = archive.get("source", {}) source_id = str(source.get("id") or source.get("key") or "site") kind = archive["kind"] @@ -692,7 +697,7 @@ def store_source_archive( source_system="jira_source_archive", source_identifier=source_id, source_uri=source.get("self"), - blob_key=f"jira-source/{scope}/{kind}/{source_id}/{digest}.json.gz", + blob_key=f"jira-source/{scope}/{kind}/{source_id}/{digest}/{blob_digest}.json.gz", metadata={ "schema": archive["schema"], "content_sha256": digest, @@ -712,7 +717,7 @@ def store_source_archive( return { "file_copy_concept_id": receipt["concept_id"], "content_sha256": digest, - "blob_sha256": hashlib.sha256(data).hexdigest(), + "blob_sha256": blob_digest, "size_bytes": len(data), "complete": bool(archive["complete"]), "schema": archive["schema"], diff --git a/src/backend/services/task_dispatch_authority_service.py b/src/backend/services/task_dispatch_authority_service.py new file mode 100644 index 00000000..006eef87 --- /dev/null +++ b/src/backend/services/task_dispatch_authority_service.py @@ -0,0 +1,72 @@ +"""Current actor authority for deliberate coding-task assignment. + +Historical creators and copied assignments are data. These operational receipts +are never exported with a project and do not replace task visibility or home +checks. Only the canonical assignment service records them. +""" + +from __future__ import annotations + +from datetime import UTC, datetime +import uuid + +from ..db.mongo_client import get_db +from ..security.access_control import get_effective_user_concept_id + +COLLECTION = "task_dispatch_authorities" + + +def _collection(): + database = get_db() + if database is None: + raise RuntimeError("Task dispatch authority store unavailable") + return database[COLLECTION] + + +def record_assignment(task: dict) -> dict | None: + actor = get_effective_user_concept_id() + if not actor: + revoke_assignment(task["task_concept_id"]) + return None + task_id = task["task_concept_id"] + receipt = { + "_id": task_id, + "dispatch_id": uuid.uuid4().hex, + "actor_concept_id": actor, + "assignee_concept_id": task.get("assignee_concept_id"), + "project_concept_id": task.get("project_concept_id"), + "organisation_concept_id": task.get("organisation_concept_id"), + "recorded_at": datetime.now(UTC), + } + _collection().replace_one({"_id": task_id}, receipt, upsert=True) + return receipt + + +def revoke_assignment(task_id: str) -> None: + # Keep the requirement for a fresh assignment even if editable task + # metadata is replaced or its federation marker is removed. + _collection().update_one( + {"_id": task_id}, + {"$set": {"actor_concept_id": None, "revoked": True}}, + upsert=True, + ) + + +def has_assignment_record(task_id: str) -> bool: + return _collection().find_one({"_id": task_id}, {"_id": 1}) is not None + + +def is_authorised_assignment(task: dict, actor_id: str) -> bool: + receipt = _collection().find_one({"_id": task["task_concept_id"]}) + return bool( + receipt + and receipt.get("actor_concept_id") == actor_id + and all( + receipt.get(field) == task.get(field) + for field in ( + "assignee_concept_id", + "project_concept_id", + "organisation_concept_id", + ) + ) + ) diff --git a/src/backend/services/task_management_service.py b/src/backend/services/task_management_service.py index 3b121eb6..4b369ef0 100644 --- a/src/backend/services/task_management_service.py +++ b/src/backend/services/task_management_service.py @@ -1742,6 +1742,24 @@ def create_task( logger.debug("Failed to append task_created history event: %s", e) workflow_event_launch: dict[str, Any] | None = None + # A native create-and-assign by its current creator is deliberate dispatch. + # Historical/imported attribution does not supply that actor authority. + from ..security.access_control import get_effective_user_concept_id + + if ( + assignee_concept_id + and canonical_task_source_id == DEFAULT_TASK_SOURCE_ID + and created_by_concept_id + and get_effective_user_concept_id() == created_by_concept_id + ): + from .task_dispatch_authority_service import record_assignment + + if record_assignment(result): + ConceptsRepository.update_one( + {"concept_id": task_concept_id}, + {"$set": {"metadata.dispatch_requested": True}}, + ) + result["dispatch_requested"] = True # Event-driven workflow launch is best-effort and must not block task writes. try: workflow_event_launch = maybe_launch_task_created_workflow( @@ -2230,6 +2248,8 @@ def _build_task_response( "history_count": len(history), "external_references": external_references, "project_concept_id": metadata.get("project_concept_id"), + "federation": metadata.get("federation"), + "dispatch_requested": bool(metadata.get("dispatch_requested")), "collection_concept_ids": metadata.get("collection_concept_ids", []), "bulk_task_collections": bulk_task_collections, "hidden_by_default_bulk_task_collections": ( @@ -2490,7 +2510,18 @@ def assign_task( ) except Exception as e: logger.debug("Failed to append task assignment history event: %s", e) - return get_task(task_concept_id) + assigned = get_task(task_concept_id) + if update_visibility and assigned.get("assignee_concept_id") == assignee_concept_id: + from .task_dispatch_authority_service import record_assignment + + receipt = record_assignment(assigned) + if receipt: + ConceptsRepository.update_one( + {"concept_id": task_concept_id}, + {"$set": {"metadata.dispatch_requested": True}}, + ) + assigned["dispatch_requested"] = True + return assigned def get_tasks_for_user( @@ -3476,6 +3507,7 @@ def _search_tasks( updated_to: str | None = None, dependency_state: str | None = None, organisation_concept_id: str | None = None, + organisation_scope_mode: str | None = None, bulk_visibility: str | None = BULK_TASK_VISIBILITY_INCLUDE, bulk_collection_ids: list[str] | str | None = None, limit: int = 50, @@ -3493,9 +3525,10 @@ def _search_tasks( except (TypeError, ValueError): offset = 0 - query_filter: Dict[str, Any] = { - "relationships.is_an_instance_of": TASK_SPECIFICATION_TYPE_ID - } + query_filter = _build_task_listing_query( + organisation_concept_id=organisation_concept_id, + organisation_scope_mode=organisation_scope_mode, + ) if project_concept_id: query_filter["metadata.project_concept_id"] = project_concept_id if collection_concept_id: @@ -3504,14 +3537,6 @@ def _search_tasks( query_filter.setdefault("$and", []).append( collection_task_query(collection_concept_id) ) - if organisation_concept_id: - org_id = _normalise_optional_concept_id(organisation_concept_id) - if not org_id: - raise InvalidTaskDataError( - f"Invalid organisation_concept_id: {organisation_concept_id}" - ) - query_filter[f"metadata.{TASK_METADATA_KEY_ORGANISATION}"] = org_id - assignee_id = None if assignee_concept_id is not None: assignee_id = _normalise_optional_concept_id(assignee_concept_id) @@ -4030,6 +4055,14 @@ def transition_task( @guard_task_write def unassign_task(task_concept_id: str) -> Dict[str, Any]: task_concept_id, doc = _get_task_doc(task_concept_id) + if (doc.get("metadata") or {}).get("dispatch_requested"): + from .task_dispatch_authority_service import revoke_assignment + + revoke_assignment(task_concept_id) + ConceptsRepository.update_one( + {"concept_id": task_concept_id}, + {"$unset": {"metadata.dispatch_requested": ""}}, + ) current_assignees = doc.get("relationships", {}).get(PREDICATE_HAS_ASSIGNEE, []) if isinstance(current_assignees, str): current_assignees = [current_assignees] diff --git a/src/backend/services/task_project_file_transfer_service.py b/src/backend/services/task_project_file_transfer_service.py new file mode 100644 index 00000000..cf10ba34 --- /dev/null +++ b/src/backend/services/task_project_file_transfer_service.py @@ -0,0 +1,55 @@ +"""Stage snapshot-bound task files without publishing any concepts. + +The admitted snapshot supplies the exact content hashes. An existing private +recovery cache may supply bytes, but is never trusted without a fresh digest. +""" + +from __future__ import annotations + +import hashlib +from pathlib import Path + +from .task_project_transfer_service import decoded, verify_snapshot + + +def stage_project_files(envelope, *, config, origin, blob_store, source_bytes): + payload = verify_snapshot(envelope, config=config, origin=origin) + staged = {} + for document in decoded(payload["body"])["files"]: + attrs = document["attributes"] + expected = attrs["sha256"] + if len(expected) != 64 or any(c not in "0123456789abcdef" for c in expected): + raise ValueError("Invalid task source file digest") + blob_key = f"federated-task-files/{expected}" + try: + data = blob_store.get_bytes(blob_key) + except FileNotFoundError: + data = source_bytes(document) + if ( + len(data) != attrs["size_bytes"] + or hashlib.sha256(data).hexdigest() != expected + ): + raise ValueError("Task source bytes differ from the admitted snapshot") + ref = blob_store.put_bytes( + blob_key, data, content_type=attrs.get("content_type") + ) + if hashlib.sha256(blob_store.get_bytes(ref.key)).hexdigest() != expected: + raise ValueError("Destination task file read-back failed") + staged[document["concept_id"]] = { + "blob_backend": ref.backend, + "blob_key": ref.key, + "blob_uri": ref.uri, + } + return staged + + +def cached_source_bytes(cache: Path): + """Only hash-named bytes in the explicitly selected private restore cache.""" + + def read(document): + digest = document["attributes"]["sha256"] + if len(digest) != 64 or any(c not in "0123456789abcdef" for c in digest): + raise ValueError("Invalid task file digest") + return (cache / "restored-task-files" / digest).read_bytes() + + return read diff --git a/src/backend/services/task_project_handoff_service.py b/src/backend/services/task_project_handoff_service.py new file mode 100644 index 00000000..cc2ba3dc --- /dev/null +++ b/src/backend/services/task_project_handoff_service.py @@ -0,0 +1,170 @@ +"""Operator-mediated signed handoff: prepare destination, release source, activate. + +No timeout activates either side. Retried messages are bound to the same digest +and epoch. Once released, recovery proceeds at the new home; an old snapshot +cannot revoke or replace work it subsequently accepted. +""" + +from __future__ import annotations + +import hashlib +import hmac +from datetime import UTC, datetime + +from pymongo import ReturnDocument + +from ..db.mongo_client import get_db +from .knowledge_federation.protocol import encode, key, peer_config, sign +from .task_project_home_service import ( + get_project_home, + local_node_id, + transfer_frozen_home, +) +from .task_project_publication_service import RECEIPTS +from .task_project_transfer_service import capture_project, portable + +VERSION = "task_project_handoff.v1" + + +def _verify(envelope, *, config, peer, direction, kind): + settings = peer_config(config, direction, peer) + payload = envelope["payload"] + expected = hmac.new(key(settings), encode(payload), hashlib.sha256).hexdigest() + if not hmac.compare_digest(expected, envelope.get("signature", "")): + raise PermissionError("Invalid task home handoff signature") + if ( + payload.get("schema_version") != VERSION + or payload.get("kind") != kind + or payload.get("sender") != peer + or payload.get("recipient") != config["node_id"] + or payload.get("project_id") not in settings.get("task_projects", []) + ): + raise PermissionError("Task home handoff is not admitted") + return payload + + +def publication_receipt(receipt_id, *, config, origin): + database = get_db() + receipt = database[RECEIPTS].find_one({"_id": receipt_id}) + if ( + not receipt + or receipt["origin"] != origin + or receipt["recipient"] != config["node_id"] + ): + raise ValueError("Canonical publication receipt is missing") + home = get_project_home(receipt["project_id"]) + if ( + not home + or home["state"] not in {"prepared", "active"} + or home["snapshot_digest"] != receipt["snapshot_digest"] + ): + raise ValueError("Publication home differs from its receipt") + return sign( + { + "schema_version": VERSION, + "kind": "prepared", + "sender": config["node_id"], + "recipient": origin, + "project_id": receipt["project_id"], + "snapshot_digest": receipt["snapshot_digest"], + "source_epoch": receipt["source_epoch"], + "home_epoch": receipt["home_epoch"], + "home_url": home["home_url"], + "source_captured_at": home.get("source_captured_at"), + "publication_id": receipt_id, + "canonical_readback": True, + }, + key(peer_config(config, "imports", origin)), + ) + + +def release_source(envelope, *, config, destination): + receipt = _verify( + envelope, config=config, peer=destination, direction="exports", kind="prepared" + ) + if local_node_id() != config["node_id"] or not receipt.get("canonical_readback"): + raise ValueError("Source node or publication read-back is invalid") + project_id = receipt["project_id"] + home = get_project_home(project_id) + expected = receipt["source_epoch"] + if not home: + raise ValueError("Source home is absent") + if home["state"] == "frozen": + # Reconcile any intervening change rather than trusting an earlier file. + fresh = capture_project(project_id, config=config, recipient=destination)[ + "payload" + ] + if fresh["digest"] != receipt["snapshot_digest"] or home["epoch"] != expected: + raise ValueError("Source changed after destination preparation") + moved = transfer_frozen_home( + project_id=project_id, + expected_epoch=expected, + node_id=destination, + home_url=receipt["home_url"], + snapshot_digest=receipt["snapshot_digest"], + receipt=receipt, + ) + if moved["epoch"] != receipt["home_epoch"]: + raise ValueError("Destination home epoch mismatch") + return sign( + { + "schema_version": VERSION, + "kind": "released", + "sender": config["node_id"], + "recipient": destination, + "project_id": project_id, + "home_epoch": moved["epoch"], + "snapshot_digest": moved["snapshot_digest"], + "publication_id": receipt["publication_id"], + "source_read_only": True, + "canonical_readback": True, + }, + key(peer_config(config, "exports", destination)), + ) + + +def activate_destination(envelope, *, config, origin): + release = _verify( + envelope, config=config, peer=origin, direction="imports", kind="released" + ) + if ( + local_node_id() != config["node_id"] + or not release.get("source_read_only") + or not release.get("canonical_readback") + ): + raise ValueError("Source release or destination node is invalid") + database = get_db() + receipt = database[RECEIPTS].find_one({"_id": release["publication_id"]}) + if not receipt or any( + receipt.get(k) != release[k] + for k in ("project_id", "home_epoch", "snapshot_digest") + ): + raise ValueError("Source release differs from the canonical publication") + query = { + "_id": release["project_id"], + "state": {"$in": ["prepared", "active"]}, + "epoch": release["home_epoch"], + "home_node_id": config["node_id"], + "snapshot_digest": release["snapshot_digest"], + } + home = database.task_project_homes.find_one_and_update( + query, + { + "$set": { + "state": "active", + "source_release": release, + "activated_at": datetime.now(UTC), + } + }, + return_document=ReturnDocument.AFTER, + ) + if home is None: + raise ValueError("Destination home changed; activation refused") + return { + "project_id": release["project_id"], + "home_epoch": home["epoch"], + "state": home["state"], + "snapshot_digest": home["snapshot_digest"], + "canonical_readback": True, + "source_release": portable(release), + } diff --git a/src/backend/services/task_project_home_service.py b/src/backend/services/task_project_home_service.py index 386e33b3..e9c2917d 100644 --- a/src/backend/services/task_project_home_service.py +++ b/src/backend/services/task_project_home_service.py @@ -62,7 +62,6 @@ def guard_task_write(function): @wraps(function) def guarded(*args, **kwargs): from .task_management_service import _get_task_doc - from ..security.access_control import can_access_concept arguments = signature.bind(*args, **kwargs).arguments project_ids = {arguments.get("project_concept_id")} @@ -93,8 +92,6 @@ def guarded(*args, **kwargs): ) with ExitStack() as stack: for project_id in sorted(project_ids - {None, ""}): - if not can_access_concept(project_id): - raise PermissionError("Task project is not accessible") stack.enter_context(project_write(project_id)) return function(*args, **kwargs) @@ -284,6 +281,7 @@ def transfer_frozen_home( "state": "replica", "snapshot_digest": snapshot_digest, "transfer_receipt": receipt, + "source_captured_at": receipt.get("source_captured_at"), "updated_at": datetime.now(UTC), }, "$inc": {"epoch": 1}, @@ -304,3 +302,171 @@ def transfer_frozen_home( return current raise TaskProjectHomeError("task_home_transfer_conflict", current) return get_project_home(project_id) + + +def _document_project(document: dict) -> str | None: + attributes = document.get("attributes") or {} + if "task_project" in attributes: + return document.get("concept_id") + return (document.get("metadata") or {}).get("project_concept_id") or ( + attributes.get("task_collection") or {} + ).get("project_concept_id") + + +@contextmanager +def concept_mutation( + collection, + query: dict, + *, + update: dict | None = None, + document: dict | None = None, + many: bool = False, +): + """Keep generic concept/relationship writes inside the same home barrier. + + Derived embedding maintenance does not change represented task content. + This is a canonical repository boundary, not protection from arbitrary + administrator database access or a pre-upgrade process. + """ + if update: + fields = { + field + for value in update.values() + if isinstance(value, dict) + for field in value + } + if fields and fields <= { + "embedding_status", + "embedding_updated_at", + "embedding_error", + }: + yield query + return + projection = { + "concept_id": 1, + "metadata.project_concept_id": 1, + "attributes.task_project": 1, + "attributes.task_collection.project_concept_id": 1, + } + docs = ( + [document] + if document is not None + else list(collection.find(query, projection).limit(0 if many else 1)) + ) + project_ids = {_document_project(doc) for doc in docs} + if update: + changes = update.get("$set", update) + project_ids.add(changes.get("metadata.project_concept_id")) + project_ids.add((changes.get("metadata") or {}).get("project_concept_id")) + with ExitStack() as stack: + for project_id in sorted(project_ids - {None, ""}): + stack.enter_context(project_write(project_id)) + # A concurrent project move must not turn this admitted write into a + # mutation of a different, possibly already frozen home. + preimages = [ + { + "_id": doc["_id"], + "metadata.project_concept_id": (doc.get("metadata") or {}).get( + "project_concept_id" + ), + } + for doc in docs + if "_id" in doc + ] + # Preserve ordinary upsert semantics when no preimage exists. + constraint = {"$or": preimages} if preimages else None + yield ( + {"$and": [query, constraint]} if document is None and constraint else query + ) + + +@contextmanager +def text_relation_mutation( + collection, query: dict, *, document: dict | None = None, update: dict | None = None +): + """Text deletion/replacement must obey the owning task-project home too.""" + from ..db.repositories.concepts_repository import ConceptsRepository + + rows = ( + [document] + if document is not None + else list(collection.find(query, {"subject_concept_id": 1})) + ) + ids = {row.get("subject_concept_id") for row in rows} - {None} + # Upserts and subject changes have no corresponding preimage yet. + for fields in ( + query, + update or {}, + (update or {}).get("$set", {}), + (update or {}).get("$setOnInsert", {}), + ): + subject = fields.get("subject_concept_id") + if isinstance(subject, str): + ids.add(subject) + if not ids: + yield query + return + with concept_mutation( + ConceptsRepository.collection(), {"concept_id": {"$in": sorted(ids)}}, many=True + ): + preimages = [ + {"_id": row["_id"], "subject_concept_id": row.get("subject_concept_id")} + for row in rows + if "_id" in row + ] + yield ( + {"$and": [query, {"$or": preimages}]} + if document is None and preimages + else query + ) + + +def workflow_home_allows_execution(document: dict, *, database=None) -> bool: + """Home routing adds no actor authority; existing claim checks still apply.""" + inputs = document.get("inputs") or {} + task_id = inputs.get("task_concept_id") or inputs.get("task_id") + if not isinstance(task_id, str) or not task_id: + return True + if database is None: + from ..db.mongo_client import get_db + + database = get_db() + if database is None: + return False + task = database.concepts.find_one( + {"concept_id": task_id}, {"metadata.project_concept_id": 1} + ) + project_id = ((task or {}).get("metadata") or {}).get("project_concept_id") + if not project_id: + return True + home = database[COLLECTION].find_one({"_id": project_id}) + return home is None or ( + home.get("state") == "active" and home.get("home_node_id") == local_node_id() + ) + + +def project_home_summaries(project_ids: list[str]) -> dict[str, dict]: + """Bounded route/freshness projection after caller checks project visibility.""" + rows = _collection().find( + {"_id": {"$in": project_ids}}, + { + "home_node_id": 1, + "home_url": 1, + "epoch": 1, + "state": 1, + "source_captured_at": 1, + "updated_at": 1, + }, + ) + return { + row["_id"]: { + "node_id": row.get("home_node_id"), + "url": row.get("home_url"), + "state": row.get("state"), + "epoch": row.get("epoch"), + "writable_here": row.get("state") == "active" + and row.get("home_node_id") == local_node_id(), + "snapshot_at": row.get("source_captured_at"), + } + for row in rows + } diff --git a/src/backend/services/task_project_publication_service.py b/src/backend/services/task_project_publication_service.py new file mode 100644 index 00000000..054266cc --- /dev/null +++ b/src/backend/services/task_project_publication_service.py @@ -0,0 +1,442 @@ +"""Atomic publication of an admitted task-project snapshot. + +The transfer adapter owns these canonical concept/text writes. File bytes must +already be staged and verified. One MongoDB transaction publishes the complete +view and a prepared (non-executing) home. Execution state is never imported. +""" + +from __future__ import annotations + +import copy +import hashlib +from datetime import UTC, datetime + +from bson import ObjectId +from pymongo import InsertOne, ReplaceOne, UpdateOne +from pymongo.read_concern import ReadConcern +from pymongo.write_concern import WriteConcern + +from ..db.mongo_client import get_db +from ..security.access_control import get_effective_user_concept_id +from .knowledge_federation.protocol import digest +from .task_project_transfer_service import ( + decoded, + portable, + verify_snapshot, + _semantic_concept, +) +from .text_value_service import _compute_fingerprint + +RECEIPTS = "task_project_transfer_receipts" + + +def concept_digest(document: dict | None) -> str | None: + return digest(portable(_semantic_concept(document))) if document else None + + +def destination_text_digest(database, ids, *, session=None): + relations = list( + database.text_relations.find( + {"subject_concept_id": {"$in": ids}}, session=session + ) + ) + value_ids = {r["object_text_id"] for r in relations} + value_ids.update( + ObjectId(str(v)) for v in list(value_ids) if ObjectId.is_valid(str(v)) + ) + values = list( + database.text_values.find({"_id": {"$in": list(value_ids)}}, session=session) + ) + return digest( + portable( + { + "relations": sorted(relations, key=lambda r: str(r["_id"])), + "values": sorted(values, key=lambda r: str(r["_id"])), + } + ) + ) + + +def check_identity_and_retention(existing: dict, incoming: dict) -> None: + """A shared slug alone is never authority to replace an existing object.""" + old, new = existing.get("metadata", {}), incoming.get("metadata", {}) + old_jira = old.get("external_references", {}).get("jira", {}) + new_jira = new.get("external_references", {}).get("jira", {}) + if old_jira.get("issue_id") and old_jira.get("issue_id") == new_jira.get( + "issue_id" + ): + for field in ("task_history", "comments", "worklog", "attachments"): + incoming_entries = {digest(portable(row)) for row in new.get(field, [])} + if any( + digest(portable(row)) not in incoming_entries + for row in old.get(field, []) + ): + raise ValueError( + "Destination task history is not retained by the source" + ) + return + a, b = existing.get("attributes", {}), incoming.get("attributes", {}) + if a.get("sha256") and a.get("sha256") == b.get("sha256"): + return + for field, keys in ( + ("task_project", ("source_site", "source_id", "organisation_concept_id")), + ("task_collection", ("project_concept_id", "source_id")), + ): + x, y = a.get(field), b.get(field) + if x and y and all(x.get(k) == y.get(k) for k in keys): + return + if concept_digest(existing) == concept_digest(incoming): + return + raise ValueError("Unresolved destination concept identity collision") + + +def verify_staged_files(payload: dict, staged: dict, blob_store) -> dict: + """Read back actual destination bytes before opening a database transaction.""" + files = decoded(payload["body"])["files"] + if set(staged) != {r["concept_id"] for r in files}: + raise ValueError("Staged file inventory differs from the snapshot") + result = {} + for document in files: + cid = document["concept_id"] + expected = document.get("attributes", {}) + record = staged[cid] + data = blob_store.get_bytes(record["blob_key"]) + if len(data) != expected.get("size_bytes") or hashlib.sha256( + data + ).hexdigest() != expected.get("sha256"): + raise ValueError("Staged task file bytes do not match the source") + result[cid] = {k: record[k] for k in ("blob_key", "blob_uri", "blob_backend")} + return result + + +def publish_project_snapshot( + envelope: dict, + *, + config: dict, + origin: str, + expected_destination_digests: dict, + expected_destination_text_digest: str, + staged_files: dict, + blob_store, + database=None, + _after_concepts=None, +) -> dict: + """Publish data and a prepared home, preserving a prior view on interruption. + + The private callback is a fault-injection seam for transaction acceptance. + It is not exposed through a tool or HTTP route. Replaying a committed + snapshot returns its receipt without replacing later accepted writes. + """ + payload = verify_snapshot(envelope, config=config, origin=origin) + actor = get_effective_user_concept_id() + if not actor or actor != payload["actor"]: + raise PermissionError( + "Task transfer actor does not match the admitted snapshot" + ) + project_id = payload["project_id"] + home = decoded(payload.get("source_home")) + if not home or home.get("state") != "frozen" or home.get("home_node_id") != origin: + raise ValueError("A quiescent frozen source home is required") + database = database if database is not None else get_db() + if database is None or not database.client.admin.command("hello").get("setName"): + raise RuntimeError("Task project publication requires MongoDB transactions") + receipt_id = digest([origin, project_id, payload["digest"], home["epoch"]]) + prior = database[RECEIPTS].find_one({"_id": receipt_id}) + if prior: + return prior + file_locations = verify_staged_files(payload, staged_files, blob_store) + body = decoded(payload["body"]) + documents = body["concepts"] + body["files"] + ids = [r["concept_id"] for r in documents] + if set(expected_destination_digests) != set(ids): + raise ValueError("Exact destination preflight inventory is required") + now = datetime.now(UTC) + + def commit(session): + prior = database[RECEIPTS].find_one({"_id": receipt_id}, session=session) + if prior: + return prior + if database.task_project_homes.find_one({"_id": project_id}, session=session): + raise ValueError( + "Destination project already has a home; reconcile instead of replaying" + ) + if ( + destination_text_digest(database, ids, session=session) + != expected_destination_text_digest + ): + raise ValueError("Destination text changed after transfer preflight") + existing = { + r["concept_id"]: r + for r in database.concepts.find( + {"concept_id": {"$in": ids}}, session=session + ) + } + replacements = [] + for source in documents: + cid = source["concept_id"] + previous = existing.get(cid) + if concept_digest(previous) != expected_destination_digests[cid]: + raise ValueError("Destination changed after transfer preflight") + if previous: + check_identity_and_retention(previous, source) + document = copy.deepcopy(source) + if previous: + document["_id"] = previous["_id"] + if previous.get("guid"): + document["guid"] = previous["guid"] + document["embedding_status"] = "pending" + document.setdefault("metadata", {})["federation"] = { + "origin": origin, + "source_concept_id": cid, + "source_project_id": project_id, + "snapshot_digest": payload["digest"], + "transferred_at": now, + "dispatch_requires_current_actor": True, + } + if cid in file_locations: + document.setdefault("attributes", {}).update(file_locations[cid]) + document["attributes"]["file_copy_metadata_storage"] = "attributes.v1" + replacements.append(ReplaceOne({"concept_id": cid}, document, upsert=True)) + if replacements: + database.concepts.bulk_write(replacements, session=session) + task_ids = [ + document["concept_id"] + for document in body["concepts"] + if (document.get("metadata") or {}).get("project_concept_id") == project_id + ] + if database.task_dispatch_authorities.find_one( + {"_id": {"$in": task_ids}, "actor_concept_id": {"$ne": None}}, + session=session, + ): + raise ValueError( + "Destination task has a current assignment; reconcile before transfer" + ) + if task_ids: + database.task_dispatch_authorities.bulk_write( + [ + UpdateOne( + {"_id": cid}, + { + "$set": { + "actor_concept_id": None, + "revoked": True, + "imported_snapshot_digest": payload["digest"], + "project_concept_id": project_id, + } + }, + upsert=True, + ) + for cid in task_ids + ], + session=session, + ) + if _after_concepts: + _after_concepts(database, session) + + # Exact text identity preserves bytes even where older normalised + # fingerprints had collapsed whitespace/case variants. + wanted = {} + for value in body["text_values"]: + fingerprint = _compute_fingerprint( + value["text"], value.get("lang", "en"), identity_mode="exact" + ) + wanted[str(value["_id"])] = (fingerprint, value) + by_fingerprint = { + r["fingerprint"]: r + for r in database.text_values.find( + {"fingerprint": {"$in": sorted({fp for fp, _ in wanted.values()})}}, + session=session, + ) + } + inserts, value_map = [], {} + for source_id, (fingerprint, source) in wanted.items(): + value = by_fingerprint.get(fingerprint) + if value is None: + value = copy.deepcopy(source) + value["_id"] = ObjectId() + value["fingerprint"] = fingerprint + by_fingerprint[fingerprint] = value + inserts.append(InsertOne(value)) + if value["text"] != source["text"] or value.get("lang", "en") != source.get( + "lang", "en" + ): + raise ValueError("Destination exact text identity collision") + value_map[source_id] = value["_id"] + if inserts: + database.text_values.bulk_write(inserts, session=session) + database.text_relations.delete_many( + {"subject_concept_id": {"$in": ids}}, session=session + ) + relations = [] + seen = set() + for original in body["text_relations"]: + relation = copy.deepcopy(original) + source_relation_id = str(relation.pop("_id")) + source_value_id = str(relation["object_text_id"]) + relation["object_text_id"] = value_map[str(relation["object_text_id"])] + semantic_key = ( + relation["subject_concept_id"], + relation["predicate"], + str(relation["object_text_id"]), + ) + if semantic_key in seen: + raise ValueError("Source text relations collapse under exact identity") + seen.add(semantic_key) + relation["context"] = dict(relation.get("context") or {}) + relation.setdefault("context", {})[ + "federation_source_relation_id" + ] = source_relation_id + relation["context"]["federation_source_text_value"] = { + k: v + for k, v in wanted[source_value_id][1].items() + if k not in {"text", "lang", "fingerprint"} + } + relation["_id"] = ObjectId() + relations.append(InsertOne(relation)) + if relations: + database.text_relations.bulk_write(relations, session=session) + record = next(r for r in body["concepts"] if r["concept_id"] == project_id)[ + "attributes" + ]["task_project"] + prepared = { + "_id": project_id, + "schema_version": "task_project_home.v1", + "home_node_id": config["node_id"], + "home_url": config["home_url"], + "epoch": home["epoch"] + 1, + "state": "prepared", + "operations": {}, + "source_identity": home["source_identity"], + "snapshot_digest": payload["digest"], + "updated_at": now, + "source_captured_at": payload["captured_at"], + } + database.task_project_homes.insert_one(prepared, session=session) + receipt = { + "_id": receipt_id, + "schema_version": "task_project_publication.v1", + "project_id": project_id, + "origin": origin, + "recipient": config["node_id"], + "snapshot_digest": payload["digest"], + "source_epoch": home["epoch"], + "home_epoch": prepared["epoch"], + "state": "prepared", + "concept_count": len(body["concepts"]), + "file_count": len(body["files"]), + "text_relation_count": len(relations), + "published_at": now, + "canonical_readback": True, + "dispatch_state_copied": False, + } + # Read within the transaction to distinguish intended from persisted data. + if database.concepts.count_documents( + {"concept_id": {"$in": ids}}, session=session + ) != len(ids): + raise RuntimeError("Published task-project inventory is incomplete") + if database.text_relations.count_documents( + {"subject_concept_id": {"$in": ids}}, session=session + ) != len(relations): + raise RuntimeError("Published task-project text inventory is incomplete") + database[RECEIPTS].insert_one(receipt, session=session) + return receipt + + with database.client.start_session() as session: + receipt = session.with_transaction( + commit, + read_concern=ReadConcern("snapshot"), + write_concern=WriteConcern("majority"), + ) + persisted = database[RECEIPTS].find_one({"_id": receipt_id}) + if not persisted: + raise RuntimeError("Committed task-project receipt is not readable") + return persisted + + +def verify_project_publication(envelope, *, config, origin, database=None): + """Compare the complete admitted semantic slice with canonical storage. + + Local GUIDs on proven collisions, exact-text storage identities, blob routes + and local federation receipts may differ. Source text-value provenance is + retained on each relation when exact text reuses an existing local value. + """ + payload = verify_snapshot(envelope, config=config, origin=origin) + if get_effective_user_concept_id() != payload["actor"]: + raise PermissionError("Task publication verification actor mismatch") + database = database if database is not None else get_db() + body = decoded(payload["body"]) + documents = body["concepts"] + body["files"] + ids = [r["concept_id"] for r in documents] + existing = { + r["concept_id"]: r for r in database.concepts.find({"concept_id": {"$in": ids}}) + } + file_ids = {r["concept_id"] for r in body["files"]} + for document in documents: + cid = document["concept_id"] + if cid not in existing: + raise ValueError("Published task concept is missing") + source, target = _semantic_concept(document), _semantic_concept(existing[cid]) + if "metadata" not in source and not target.get("metadata"): + target.pop("metadata", None) + for value in (source, target): + value.pop("guid", None) + if cid in file_ids: + for field in ( + "blob_backend", + "blob_key", + "blob_uri", + "file_copy_metadata_storage", + ): + value.get("attributes", {}).pop(field, None) + if digest(portable(source)) != digest(portable(target)): + raise ValueError(f"Published task concept differs: {cid}") + relations = list(database.text_relations.find({"subject_concept_id": {"$in": ids}})) + source_values = {str(v["_id"]): v for v in body["text_values"]} + target_values = { + str(v["_id"]): v + for v in database.text_values.find( + {"_id": {"$in": [r["object_text_id"] for r in relations]}} + ) + } + by_source = { + r.get("context", {}).get("federation_source_relation_id"): r for r in relations + } + if len(relations) != len(body["text_relations"]) or len(by_source) != len( + relations + ): + raise ValueError("Published text relation inventory differs") + for relation in body["text_relations"]: + source = copy.deepcopy(relation) + target = copy.deepcopy(by_source.get(str(source.pop("_id")))) + if target is None: + raise ValueError("Published source text relation is missing") + target.pop("_id") + source_value = source_values[str(source.pop("object_text_id"))] + target_value = target_values[str(target.pop("object_text_id"))] + context = target["context"] + context.pop("federation_source_relation_id") + provenance = context.pop("federation_source_text_value") + if "context" not in source and not context: + target.pop("context") + expected_provenance = { + k: v + for k, v in source_value.items() + if k not in {"text", "lang", "fingerprint"} + } + if ( + digest(portable(source)) != digest(portable(target)) + or provenance != expected_provenance + or any( + source_value.get(k, "en" if k == "lang" else None) + != target_value.get(k, "en" if k == "lang" else None) + for k in ("text", "lang") + ) + ): + raise ValueError("Published task text or provenance differs") + return { + "project_id": payload["project_id"], + "snapshot_digest": payload["digest"], + "concepts_verified": len(documents), + "text_relations_verified": len(relations), + "canonical_readback": True, + } diff --git a/src/backend/services/task_project_service.py b/src/backend/services/task_project_service.py index e4f1b72b..7b328ad1 100644 --- a/src/backend/services/task_project_service.py +++ b/src/backend/services/task_project_service.py @@ -14,6 +14,7 @@ from ..db.repositories.concepts_repository import ConceptsRepository from ..security.access_control import can_access_concept +from .task_project_home_service import project_home_summaries from .concept_service import ( ConceptNotFoundError, create_concept, @@ -57,7 +58,36 @@ def _record(concept_id, key): def get_task_project(project_concept_id): - return _record(project_concept_id, PROJECT_RECORD) + result = _record(project_concept_id, PROJECT_RECORD) + home = project_home_summaries([project_concept_id]).get(project_concept_id) + if home: + result["write_home"] = home + return result + + +def task_execution_home(task_concept_id): + """Return only the routing facts needed by an actor who can read this task. + + Assignment may grant access to one private-project task. It must neither + require nor grant visibility of the whole project's content. + """ + task = ConceptsRepository.find_one( + {"concept_id": task_concept_id}, projection={"metadata.project_concept_id": 1} + ) + if not task: + raise ValueError("Task is not accessible") + project_id = (task.get("metadata") or {}).get("project_concept_id") + if not project_id: + return None + project = ConceptsRepository.collection().find_one( + {"concept_id": project_id}, projection={"attributes.task_project.writer": 1} + ) + return { + "writer": ((project or {}).get("attributes") or {}) + .get("task_project", {}) + .get("writer"), + "write_home": project_home_summaries([project_id]).get(project_id), + } def get_task_collection(collection_concept_id): @@ -93,11 +123,17 @@ def list_task_projects(*, limit=100, offset=0): limit=max(1, min(int(limit), 200)), skip=max(0, int(offset)), ) - return [ + result = [ {"concept_id": doc["concept_id"], **doc["attributes"][PROJECT_RECORD]} for doc in docs ] + homes = project_home_summaries([row["concept_id"] for row in result]) + for row in result: + if row["concept_id"] in homes: + row["write_home"] = homes[row["concept_id"]] + return result + def ensure_jira_project( project, diff --git a/src/backend/services/task_project_transfer_service.py b/src/backend/services/task_project_transfer_service.py index 82cbc4df..810e13a9 100644 --- a/src/backend/services/task_project_transfer_service.py +++ b/src/backend/services/task_project_transfer_service.py @@ -18,7 +18,11 @@ TextRelationsRepository, TextValuesRepository, ) -from ..security.access_control import can_access_concept, get_effective_user_concept_id +from ..security.access_control import ( + can_access_concept, + filter_accessible_concept_ids, + get_effective_user_concept_id, +) from ..security.visibility_predicates import ( get_specific_to_org_values, get_specific_to_user_values, @@ -31,6 +35,7 @@ MAX_BYTES = 256 * 1024 * 1024 # Dispatch, supervision and execution state must be established at the new home. EXECUTION_METADATA = { + "dispatch_requested", "coding_supervision", "coding_dispatch", "dispatch_authority", @@ -117,6 +122,7 @@ def capture_project(project_id: str, *, config: dict, recipient: str) -> dict: raise PermissionError("Task project export is not admitted") if not can_access_concept(project_id): raise PermissionError("Task project is inaccessible") + initial_home = get_project_home(project_id) collection = ConceptsRepository.collection() project = collection.find_one({"concept_id": project_id}) record = ((project or {}).get("attributes") or {}).get("task_project") @@ -136,9 +142,9 @@ def capture_project(project_id: str, *, config: dict, recipient: str) -> dict: if not set(collection_ids).issubset(ids): raise ValueError("Project collection inventory is incomplete") audiences = set(settings.get("audiences", [])) + if filter_accessible_concept_ids(ids) != ids: + raise PermissionError("Project contains an inaccessible record") for document in documents: - if not can_access_concept(document["concept_id"]): - raise PermissionError("Project contains an inaccessible record") scope = _scope(document) keys = {f"user:{v}" for v in scope["users"]} | { f"org:{v}" for v in scope["organisations"] @@ -146,14 +152,17 @@ def capture_project(project_id: str, *, config: dict, recipient: str) -> dict: if not keys or not keys.intersection(audiences): raise PermissionError("Task record has no admitted private audience") semantic = [_semantic_concept(d) for d in documents] - files = [] - for file_id in sorted(referenced_file_ids(semantic)): - if not can_access_concept(file_id): - raise PermissionError("A task source file is inaccessible") - document = collection.find_one({"concept_id": file_id}) - if not document: - raise ValueError("Missing task source file") - files.append(_semantic_concept(document)) + file_ids = referenced_file_ids(semantic) + if len(file_ids) > MAX_CONCEPTS: + raise ValueError("Task project exceeds the bounded file inventory") + if filter_accessible_concept_ids(file_ids) != file_ids: + raise PermissionError("A task source file is inaccessible") + file_documents = list(collection.find({"concept_id": {"$in": sorted(file_ids)}})) + if {d["concept_id"] for d in file_documents} != file_ids: + raise ValueError("Missing task source file") + files = sorted( + (_semantic_concept(d) for d in file_documents), key=lambda d: d["concept_id"] + ) relations = list( TextRelationsRepository.collection().find( { @@ -187,6 +196,8 @@ def capture_project(project_id: str, *, config: dict, recipient: str) -> dict: if len(encode(body)) > MAX_BYTES: raise ValueError("Task project exceeds the bounded snapshot size") home = get_project_home(project_id) + if home != initial_home: + raise ValueError("Project home changed during snapshot capture; retry") payload = { "schema_version": VERSION, "origin": config["node_id"], @@ -243,6 +254,22 @@ def verify_snapshot(envelope: dict, *, config: dict, origin: str) -> dict: elif record.get("metadata", {}).get("project_concept_id") != project_id: raise ValueError("Unrelated concept in task project snapshot") subject_ids = {r["concept_id"] for r in [*records, *body["files"]]} + if ( + len(subject_ids) != len(records) + len(body["files"]) + or len(body["files"]) > MAX_CONCEPTS + ): + raise ValueError("Duplicate or excessive snapshot concept identifiers") + if not collection_ids.issubset({r["concept_id"] for r in records}): + raise ValueError("Incomplete task collection inventory") + values = decoded(body["text_values"]) + value_ids = {str(r["_id"]) for r in values} + relations = decoded(body["text_relations"]) + if len(value_ids) != len(values) or value_ids != { + str(r["object_text_id"]) for r in relations + }: + raise ValueError("Incomplete or duplicate task text value inventory") + if len({str(r["_id"]) for r in relations}) != len(relations): + raise ValueError("Duplicate task text relation identifiers") if any(r["subject_concept_id"] not in subject_ids for r in body["text_relations"]): raise ValueError("Unrelated task text relation") if {r["concept_id"] for r in body["files"]} != referenced_file_ids(records): diff --git a/src/backend/workflows/durable/instance_manager.py b/src/backend/workflows/durable/instance_manager.py index 89dcbb13..82544519 100644 --- a/src/backend/workflows/durable/instance_manager.py +++ b/src/backend/workflows/durable/instance_manager.py @@ -419,7 +419,15 @@ def is_current_claim(self, instance_id: str, worker_id: str, token: str) -> bool require_unexpired=True, ) ) - return bool(doc and doc.get("task_ownership_active")) + from ...services.task_project_home_service import workflow_home_allows_execution + + return bool( + doc + and doc.get("task_ownership_active") + and workflow_home_allows_execution( + doc, database=getattr(coll, "database", None) + ) + ) def begin_claim_effect(self, instance_id, worker_id, token, method_name, payload): coll = self._get_instances_collection() @@ -1858,6 +1866,15 @@ def _claim_matching_instance( if candidate is None: return None identifier = candidate["instance_id"] + from ...services.task_project_home_service import ( + workflow_home_allows_execution, + ) + + if not workflow_home_allows_execution( + candidate, database=getattr(coll, "database", None) + ): + excluded.append(identifier) + continue key = task_key(candidate) try: claimed = coll.find_one_and_update( diff --git a/src/backend/workflows/durable/task_ownership.py b/src/backend/workflows/durable/task_ownership.py index 3d25439d..0e3db94f 100644 --- a/src/backend/workflows/durable/task_ownership.py +++ b/src/backend/workflows/durable/task_ownership.py @@ -116,7 +116,7 @@ def invoke(**payload: Any): return invoke -def claim_admission_validator(worker_pattern: str = "^worker_") -> dict[str, Any]: +def claim_admission_validator(worker_pattern: str = "^worker_", *, required_capability: str = TASK_OWNERSHIP_CAPABILITY) -> dict[str, Any]: """Database-enforced compatibility fence for legacy claim implementations. This rejects the observed old code which replaces claimed_by_build on each @@ -128,7 +128,7 @@ def claim_admission_validator(worker_pattern: str = "^worker_") -> dict[str, Any { "status": "running", "locked_by": {"$regex": worker_pattern}, - "claimed_by_build.capabilities": {"$ne": TASK_OWNERSHIP_CAPABILITY}, + "claimed_by_build.capabilities": {"$ne": required_capability}, } ] } diff --git a/src/backend/workflows/durable/worker_identity.py b/src/backend/workflows/durable/worker_identity.py index 0dee0ffb..8c87b44e 100644 --- a/src/backend/workflows/durable/worker_identity.py +++ b/src/backend/workflows/durable/worker_identity.py @@ -164,6 +164,7 @@ def build_worker_build_identity(worker_id: str) -> dict[str, Any]: version_info["capabilities"] = [ EXACT_AUTHORITY_SNAPSHOT_WORKER_CAPABILITY, "durable_task_ownership_v1", + "task_project_home_v1", ] return normalise_worker_build_identity(version_info, worker_id=worker_id) diff --git a/src/frontend/web/von_interface/static/js/components/taskPanel.js b/src/frontend/web/von_interface/static/js/components/taskPanel.js index c38b7490..cd5e7936 100644 --- a/src/frontend/web/von_interface/static/js/components/taskPanel.js +++ b/src/frontend/web/von_interface/static/js/components/taskPanel.js @@ -399,6 +399,21 @@ function getHiddenBulkTaskCollections(task) { )); } +function projectHomeNotice(project) { + const home = project?.write_home; + if (!home || home.writable_here) return ''; + const label = home.state === 'replica' ? 'Read-only project copy.' : 'Project transfer in progress.'; + const freshness = home.snapshot_at ? ` Snapshot: ${escapeHtml(String(home.snapshot_at))}.` : ''; + let route = ''; + try { + const url = new URL(home.url); + if (['https:', 'http:'].includes(url.protocol)) { + route = ` Open the project’s write home`; + } + } catch (_) { /* A missing route must not break project navigation. */ } + return `${label}${freshness}${route}`; +} + function normaliseBulkTaskCollectionSummary(collection) { if (!collection || typeof collection !== 'object') return null; const collectionId = typeof collection.collection_id === 'string' @@ -1081,7 +1096,7 @@ function renderGlobalTasksTabContent() {
-
${escapeHtml(_taskProjects.find((project) => project.concept_id === _selectedProjectId)?.description || '')}
+
${escapeHtml(_taskProjects.find((project) => project.concept_id === _selectedProjectId)?.description || '')} ${projectHomeNotice(_taskProjects.find((project) => project.concept_id === _selectedProjectId))}
@@ -1170,6 +1185,9 @@ function renderGlobalTasksTabContent() { const details = await getJson(`/api/tasks/projects/${encodeURIComponent(_selectedProjectId)}`); if (generation !== _globalTaskLoadGeneration) return; _projectCollections = details.collections || []; + if (details.project) { + _taskProjects = _taskProjects.map((project) => project.concept_id === _selectedProjectId ? details.project : project); + } } catch (error) { if (generation !== _globalTaskLoadGeneration) return; showToast('Could not load project collections', 'error'); @@ -1235,6 +1253,11 @@ function renderGlobalTasksTabContent() { if (createBtn) { createBtn.addEventListener('click', handleGlobalCreateTask); } + const selectedHome = _taskProjects.find((project) => project.concept_id === _selectedProjectId)?.write_home; + if (selectedHome && !selectedHome.writable_here) { + _globalTasksContainer.querySelectorAll('.global-tasks-create input, .global-tasks-create textarea, .global-tasks-create select, .global-tasks-create button') + .forEach((control) => { control.disabled = true; }); + } // Update the task list element reference for global mode _taskListEl = _globalTasksContainer.querySelector('#globalTaskList'); diff --git a/tests/backend/test_codex_von_worker.py b/tests/backend/test_codex_von_worker.py index ea785d2d..824271f0 100644 --- a/tests/backend/test_codex_von_worker.py +++ b/tests/backend/test_codex_von_worker.py @@ -46,6 +46,13 @@ def check_objects(schema): @pytest.fixture(autouse=True) def isolated_archive_boundary(monkeypatch): + import mongomock + from src.backend.services import task_dispatch_authority_service + + collection = mongomock.MongoClient().test.task_dispatch_authorities + monkeypatch.setattr( + task_dispatch_authority_service, "_collection", lambda: collection + ) # Existing controller tests isolate publication; canonical archive capture # and the real launch path are exercised in test_task_run_archives.py. monkeypatch.setattr(worker, "archive_checkpoint", lambda *a, **kw: True) @@ -223,12 +230,14 @@ def run(args, **kwargs): @pytest.mark.parametrize("writer", ["jira", "von", None]) def test_project_tracking_authority_is_read_live(config, monkeypatch, writer): - from src.backend.services import task_project_service + from src.backend.services import task_project_service, task_project_home_service + + monkeypatch.setattr(task_project_home_service, "get_project_home", lambda pid: None) api = object.__new__(worker.Von) api.config = config monkeypatch.setattr( - task_project_service, "get_task_project", lambda pid: {"writer": writer} + task_project_service, "task_execution_home", lambda pid: {"writer": writer} ) assert api.native_writer(task(config, project_concept_id="#V#project")) is ( writer == "von" diff --git a/tests/backend/test_concepts_repository_embedding_queue.py b/tests/backend/test_concepts_repository_embedding_queue.py index 78645539..04389411 100644 --- a/tests/backend/test_concepts_repository_embedding_queue.py +++ b/tests/backend/test_concepts_repository_embedding_queue.py @@ -4,6 +4,7 @@ from typing import Any import pytest +import mongomock from src.backend.db.repositories import concepts_repository as repository_module from src.backend.db.repositories.concepts_repository import ConceptsRepository @@ -16,9 +17,14 @@ class _WriteResult: class _FakeConceptsCollection: def __init__(self) -> None: + self.reads = mongomock.MongoClient().test.concepts + self.reads.insert_many([{"concept_id": "#V#example"}, {"concept_id": "#V#message"}]) self.inserted: list[dict[str, Any]] = [] self.updated: list[tuple[str, dict[str, Any], dict[str, Any]]] = [] + def find(self, query, projection=None): + return self.reads.find(query, projection) + def insert_one(self, document: dict[str, Any]) -> _WriteResult: self.inserted.append(document) return _WriteResult() diff --git a/tests/backend/test_jira_source_retention_service.py b/tests/backend/test_jira_source_retention_service.py index 8232f5f8..9a950a71 100644 --- a/tests/backend/test_jira_source_retention_service.py +++ b/tests/backend/test_jira_source_retention_service.py @@ -242,3 +242,35 @@ def test_archive_read_requires_access_and_verifies_bytes(monkeypatch): retention.read_source_archive( {**ref, "blob_sha256": "wrong"}, actor_concept_id="#V#owner" ) + + +def test_archive_blob_identity_is_independent_of_platform_gzip_header(monkeypatch): + from src.backend.services import computer_file_copy_service as copies + + raw_compress = gzip.compress + calls = [] + monkeypatch.setattr( + copies, + "import_bytes_file_copy", + lambda **kwargs: calls.append(kwargs) + or {"success": True, "concept_id": "#V#archive"}, + ) + archive = { + "schema": "jira_source_archive.v1", + "kind": "project", + "complete": True, + "source": {"id": "42"}, + } + for os_byte in (3, 255): + + def compress(data, *, mtime, value=os_byte): + packed = raw_compress(data, mtime=mtime) + return packed[:9] + bytes([value]) + packed[10:] + + monkeypatch.setattr(retention.gzip, "compress", compress) + retention.store_source_archive(archive, actor_concept_id="#V#owner") + assert calls[0]["data"] == calls[1]["data"] + assert calls[0]["blob_key"] == calls[1]["blob_key"] + assert calls[0]["blob_key"].endswith( + hashlib.sha256(calls[0]["data"]).hexdigest() + ".json.gz" + ) diff --git a/tests/backend/test_task_dispatch_authority_service.py b/tests/backend/test_task_dispatch_authority_service.py new file mode 100644 index 00000000..b942d67c --- /dev/null +++ b/tests/backend/test_task_dispatch_authority_service.py @@ -0,0 +1,76 @@ +import mongomock +import pytest + +from scripts import codex_von_worker as worker +from src.backend.services import task_dispatch_authority_service as authority +from src.backend.services import task_project_home_service as homes +from src.backend.services import task_project_service as projects + + +@pytest.mark.parametrize("organisation", [None, "#V#org"]) +def test_transferred_assignment_needs_current_actor_receipt_without_rewriting_creator( + monkeypatch, organisation +): + collection = mongomock.MongoClient().test.receipts + monkeypatch.setattr(authority, "_collection", lambda: collection) + config = { + "agent_id": "#V#agent", + "delegator_id": "#V#owner", + "organisation_id": "#V#org", + } + task = { + "task_concept_id": "#V#task", + "assignee_concept_id": "#V#agent", + "created_by_concept_id": "#V#historical_author", + "organisation_concept_id": "#V#org", + "project_concept_id": "#V#project", + "federation": {"origin": "source"}, + } + task["organisation_concept_id"] = organisation + assert not worker.authorised_task(task, config) + task["dispatch_requested"] = True + assert not worker.authorised_task(task, config) + monkeypatch.setattr(authority, "get_effective_user_concept_id", lambda: "#V#other") + authority.record_assignment(task) + assert not worker.authorised_task(task, config) + monkeypatch.setattr(authority, "get_effective_user_concept_id", lambda: "#V#owner") + authority.record_assignment(task) + assert worker.authorised_task(task, config) + assert task["created_by_concept_id"] == "#V#historical_author" + authority.revoke_assignment(task["task_concept_id"]) + assert not worker.authorised_task(task, config) + # Even a copied task authored by the current owner needs a new assignment. + task["created_by_concept_id"] = "#V#owner" + task["dispatch_requested"] = False + assert not worker.authorised_task(task, config) + task.pop("federation") + task.pop("project_concept_id") + assert not worker.authorised_task(task, config) + + +@pytest.mark.parametrize( + "state,node,allowed", + [ + ("prepared", "dgx", False), + ("replica", "dgx", False), + ("active", "atlas", False), + ("active", "dgx", True), + ], +) +def test_only_active_local_home_can_dispatch(monkeypatch, state, node, allowed): + monkeypatch.setenv("VON_TASK_HOME_NODE_ID", "dgx") + monkeypatch.setattr( + projects, + "task_execution_home", + lambda pid: { + "writer": "von", + "write_home": {"writable_here": state == "active" and node == "dgx"}, + }, + ) + api = object.__new__(worker.Von) + assert ( + api.native_writer( + {"task_concept_id": "#V#task", "project_concept_id": "#V#project"} + ) + is allowed + ) diff --git a/tests/backend/test_task_management_service.py b/tests/backend/test_task_management_service.py index 2090d882..0f9be66a 100644 --- a/tests/backend/test_task_management_service.py +++ b/tests/backend/test_task_management_service.py @@ -640,12 +640,17 @@ def test_update_status_invalid(self) -> None: class TestAssignTask: """Tests for assign_task function.""" + @patch( + "src.backend.services.task_dispatch_authority_service.record_assignment", + return_value=None, + ) @patch("src.backend.services.task_management_service.get_task") @patch("src.backend.services.task_management_service.ConceptsRepository") def test_assign_task_success( self, mock_repo: MagicMock, mock_get_task: MagicMock, + mock_record_assignment: MagicMock, ) -> None: """assign_task() should update assignee relationship.""" mock_repo.find_one.return_value = { diff --git a/tests/backend/test_task_project_handoff_service.py b/tests/backend/test_task_project_handoff_service.py new file mode 100644 index 00000000..38293b70 --- /dev/null +++ b/tests/backend/test_task_project_handoff_service.py @@ -0,0 +1,125 @@ +import mongomock +import pytest +from src.backend.services import task_project_handoff_service as handoff +from src.backend.services import task_project_home_service as homes +from src.backend.services.knowledge_federation.protocol import sign + +PROJECT = "#V#project" + + +@pytest.fixture +def setup(tmp_path, monkeypatch): + keyfile = tmp_path / "key" + keyfile.write_text("01" * 32) + keyfile.chmod(0o600) + settings = {"enabled": True, "key_file": str(keyfile), "task_projects": [PROJECT]} + source = {"node_id": "source", "exports": {"destination": settings}} + destination = {"node_id": "destination", "imports": {"source": settings}} + db = mongomock.MongoClient().test + monkeypatch.setattr(handoff, "get_db", lambda: db) + monkeypatch.setattr(homes, "_collection", lambda: db.task_project_homes) + return db, source, destination + + +def test_source_must_release_before_activation_and_replay_cannot_roll_back( + setup, monkeypatch +): + db, source, destination = setup + db.task_project_homes.insert_one( + { + "_id": PROJECT, + "state": "prepared", + "home_node_id": "destination", + "epoch": 2, + "snapshot_digest": "a" * 64, + "home_url": "https://destination.invalid", + "operations": {}, + } + ) + receipt = { + "_id": "receipt", + "project_id": PROJECT, + "home_epoch": 2, + "source_epoch": 1, + "snapshot_digest": "a" * 64, + "origin": "source", + "recipient": "destination", + } + db.task_project_transfer_receipts.insert_one(receipt) + monkeypatch.setenv("VON_TASK_HOME_NODE_ID", "destination") + prepared = handoff.publication_receipt( + "receipt", config=destination, origin="source" + ) + with pytest.raises(PermissionError): + handoff.activate_destination(prepared, config=destination, origin="source") + release = { + "schema_version": handoff.VERSION, + "kind": "released", + "sender": "source", + "recipient": "destination", + "project_id": PROJECT, + "home_epoch": 2, + "snapshot_digest": "a" * 64, + "publication_id": "receipt", + "source_read_only": True, + "canonical_readback": True, + } + signed = sign(release, bytes.fromhex("01" * 32)) + assert ( + handoff.activate_destination(signed, config=destination, origin="source")[ + "state" + ] + == "active" + ) + db.task_project_homes.update_one( + {"_id": PROJECT}, {"$set": {"epoch": 3, "state": "frozen"}} + ) + with pytest.raises(ValueError, match="changed"): + handoff.activate_destination(signed, config=destination, origin="source") + assert db.task_project_homes.find_one({})["epoch"] == 3 + + +def test_release_reconciles_frozen_source_against_destination_digest( + setup, monkeypatch +): + db, source, destination = setup + monkeypatch.setenv("VON_TASK_HOME_NODE_ID", "source") + db.task_project_homes.insert_one( + { + "_id": PROJECT, + "home_node_id": "source", + "state": "frozen", + "epoch": 1, + "operations": {}, + } + ) + prepared = { + "schema_version": handoff.VERSION, + "kind": "prepared", + "sender": "destination", + "recipient": "source", + "project_id": PROJECT, + "source_epoch": 1, + "home_epoch": 2, + "snapshot_digest": "a" * 64, + "home_url": "https://destination.invalid", + "publication_id": "receipt", + "canonical_readback": True, + } + envelope = sign(prepared, bytes.fromhex("01" * 32)) + monkeypatch.setattr( + handoff, "capture_project", lambda *a, **k: {"payload": {"digest": "b" * 64}} + ) + with pytest.raises(ValueError, match="changed"): + handoff.release_source(envelope, config=source, destination="destination") + assert db.task_project_homes.find_one({})["state"] == "frozen" + monkeypatch.setattr( + handoff, "capture_project", lambda *a, **k: {"payload": {"digest": "a" * 64}} + ) + release = handoff.release_source(envelope, config=source, destination="destination") + assert release["payload"]["source_read_only"] + assert ( + handoff.release_source(envelope, config=source, destination="destination") + == release + ) + assert db.task_project_homes.find_one({})["state"] == "replica" diff --git a/tests/backend/test_task_project_home_service.py b/tests/backend/test_task_project_home_service.py index ce774ebb..f99c9b97 100644 --- a/tests/backend/test_task_project_home_service.py +++ b/tests/backend/test_task_project_home_service.py @@ -108,3 +108,88 @@ def test_source_identity_collision_is_not_an_overwrite(store): evidence={"checked": True}, ) assert store.find_one({})["source_identity"]["site"] == "jira.test" + + +def test_generic_concept_and_text_writes_obey_the_same_home(store, monkeypatch): + from src.backend.db.repositories.concepts_repository import ConceptsRepository + from src.backend.db.repositories.text_value_repository import ( + TextRelationsRepository, + ) + + database = store.database + monkeypatch.setattr(ConceptsRepository, "collection", lambda: database.concepts) + monkeypatch.setattr( + TextRelationsRepository, "collection", lambda: database.text_relations + ) + database.concepts.insert_one( + {"concept_id": "#V#task", "metadata": {"project_concept_id": "#V#project"}} + ) + relation_id = database.text_relations.insert_one( + { + "subject_concept_id": "#V#task", + "predicate": "hasName", + "object_text_id": "before", + } + ).inserted_id + homes.freeze_project_home( + project_id="#V#project", expected_epoch=1, evidence={"checked": True} + ) + with pytest.raises(homes.TaskProjectHomeError): + ConceptsRepository.update_one( + {"concept_id": "#V#task"}, + {"$set": {"metadata.project_concept_id": "#V#elsewhere"}}, + ) + with pytest.raises(homes.TaskProjectHomeError): + TextRelationsRepository.delete_one({"_id": relation_id}) + with pytest.raises(homes.TaskProjectHomeError): + TextRelationsRepository.update_one( + {"subject_concept_id": "#V#task", "predicate": "hasDescription"}, + {"$set": {"object_text_id": "new"}}, + upsert=True, + ) + assert database.text_relations.find_one({"_id": relation_id}) + assert ( + database.concepts.find_one({"concept_id": "#V#task"})["metadata"][ + "project_concept_id" + ] + == "#V#project" + ) + # Derived indexing remains available on a read-only replica. + ConceptsRepository.update_one( + {"concept_id": "#V#task"}, {"$set": {"embedding_status": "indexed"}} + ) + assert ( + database.concepts.find_one({"concept_id": "#V#task"})["embedding_status"] + == "indexed" + ) + + +def test_workflow_actions_recheck_the_home_after_claim(store, monkeypatch): + store.database.concepts.insert_one( + {"concept_id": "#V#task", "metadata": {"project_concept_id": "#V#project"}} + ) + workflow = {"inputs": {"task_concept_id": "#V#task"}} + assert homes.workflow_home_allows_execution(workflow, database=store.database) + homes.freeze_project_home( + project_id="#V#project", expected_epoch=1, evidence={"checked": True} + ) + assert not homes.workflow_home_allows_execution(workflow, database=store.database) + assert homes.workflow_home_allows_execution({"inputs": {}}, database=store.database) + + +def test_a_concurrent_project_move_defeats_the_repository_write_preimage(store): + concepts = store.database.concepts + concepts.insert_one( + {"concept_id": "#V#task", "metadata": {"project_concept_id": "#V#project"}} + ) + with homes.concept_mutation(concepts, {"concept_id": "#V#task"}) as query: + concepts.update_one( + {"concept_id": "#V#task"}, + {"$set": {"metadata.project_concept_id": "#V#other"}}, + ) + assert ( + concepts.update_one( + query, {"$set": {"metadata.notes": "stale writer"}} + ).matched_count + == 0 + ) diff --git a/tests/backend/test_task_project_publication_transaction.py b/tests/backend/test_task_project_publication_transaction.py new file mode 100644 index 00000000..f46e2337 --- /dev/null +++ b/tests/backend/test_task_project_publication_transaction.py @@ -0,0 +1,226 @@ +"""Real Mongo acceptance for the project publication boundary. + +Run only against a disposable transaction fixture, never the application DB: +VON_TASK_TRANSFER_TEST_URI=mongodb://127.0.0.1:27019/?replicaSet=von_task_transfer_test +""" + +import os +import uuid +from datetime import UTC, datetime + +import pytest +from bson import ObjectId +from pymongo import MongoClient + +from src.backend.security.access_control import override_current_actor +from src.backend.services import task_project_publication_service as publish +from src.backend.services.knowledge_federation.protocol import digest, sign +from src.backend.services.task_project_transfer_service import VERSION, portable + +pytestmark = pytest.mark.skipif( + not os.getenv("VON_TASK_TRANSFER_TEST_URI"), + reason="Disposable Mongo replica-set fixture required", +) +USER = "#V#transfer_fixture_owner" +PROJECT = "#V#transfer_fixture_project" +TASK = "#V#transfer_fixture_task" + + +@pytest.fixture +def scenario(tmp_path): + uri = os.environ["VON_TASK_TRANSFER_TEST_URI"] + client = MongoClient(uri, serverSelectionTimeoutMS=5000) + assert client.admin.command("hello").get("setName") == "von_task_transfer_test" + database = client["test_project_publication_" + uuid.uuid4().hex] + database.concepts.create_index("concept_id", unique=True) + database.text_values.create_index([("fingerprint", 1), ("lang", 1)], unique=True) + database.text_relations.create_index( + [("subject_concept_id", 1), ("predicate", 1), ("object_text_id", 1)], + unique=True, + ) + old = { + "concept_id": TASK, + "metadata": { + "external_references": {"jira": {"issue_id": "42"}}, + "comments": [{"id": "old", "body": "Retained"}], + }, + "relationships": {"#V#specific_to_user": [USER]}, + } + database.concepts.insert_one(old) + database.unrelated.insert_one({"_id": "keep", "text": "Existing DGX work"}) + value_id, relation_id = ObjectId(), ObjectId() + body = portable( + { + "concepts": [ + { + "concept_id": PROJECT, + "relationships": {"#V#specific_to_user": [USER]}, + "attributes": { + "task_project": {"writer": "von", "collection_concept_ids": []} + }, + }, + { + "concept_id": TASK, + "relationships": { + "#V#specific_to_user": [USER], + "#V#has_created_by": ["#V#historical_author"], + }, + "metadata": { + "project_concept_id": PROJECT, + "external_references": {"jira": {"issue_id": "42"}}, + "comments": [ + {"id": "old", "body": "Retained"}, + {"id": "new", "body": "New source history"}, + ], + }, + }, + ], + "files": [], + "text_values": [ + { + "_id": value_id, + "text": "Original exact text", + "lang": "en", + "fingerprint": "old", + } + ], + "text_relations": [ + { + "_id": relation_id, + "subject_concept_id": TASK, + "predicate": "hasDescription", + "object_text_id": value_id, + "context": {}, + } + ], + } + ) + payload = { + "schema_version": VERSION, + "origin": "source", + "recipient": "destination", + "project_id": PROJECT, + "actor": USER, + "captured_at": datetime.now(UTC).isoformat(), + "source_home": { + "state": "frozen", + "home_node_id": "source", + "epoch": 1, + "source_identity": {"project": "42"}, + }, + "body": body, + "digest": digest(body), + } + keyfile = tmp_path / "key" + keyfile.write_text("01" * 32) + keyfile.chmod(0o600) + config = { + "node_id": "destination", + "home_url": "https://destination.invalid", + "imports": { + "source": { + "enabled": True, + "key_file": str(keyfile), + "task_projects": [PROJECT], + "identity_map": {USER: USER}, + "audiences": ["user:" + USER], + } + }, + } + args = { + "config": config, + "origin": "source", + "database": database, + "expected_destination_digests": { + PROJECT: None, + TASK: publish.concept_digest(old), + }, + "expected_destination_text_digest": publish.destination_text_digest( + database, [PROJECT, TASK] + ), + "staged_files": {}, + "blob_store": None, + } + try: + with override_current_actor(USER, None): + yield database, sign(payload, bytes.fromhex("01" * 32)), args + finally: + client.drop_database(database.name) + client.close() + + +def test_interruption_preserves_prior_complete_view_then_retry_is_atomic(scenario): + database, envelope, args = scenario + before = database.concepts.find_one({"concept_id": TASK}) + + def interrupt(db, session): + assert db.concepts.find_one({"concept_id": PROJECT}, session=session) + assert db.concepts.find_one({"concept_id": PROJECT}) is None + assert db.concepts.find_one({"concept_id": TASK}) == before + raise RuntimeError("Injected interrupted publication") + + with pytest.raises(RuntimeError, match="Injected"): + publish.publish_project_snapshot(envelope, **args, _after_concepts=interrupt) + assert database.concepts.find_one({"concept_id": TASK}) == before + assert database.concepts.find_one({"concept_id": PROJECT}) is None + assert database.task_project_homes.count_documents({}) == 0 + receipt = publish.publish_project_snapshot(envelope, **args) + assert receipt["canonical_readback"] and receipt["state"] == "prepared" + verified = publish.verify_project_publication( + envelope, config=args["config"], origin="source", database=database + ) + assert verified["text_relations_verified"] == 1 + assert database.concepts.find_one({"concept_id": TASK})["relationships"][ + "#V#has_created_by" + ] == ["#V#historical_author"] + assert database.unrelated.find_one({"_id": "keep"})["text"] == "Existing DGX work" + database.concepts.update_one( + {"concept_id": TASK}, {"$set": {"metadata.later_accepted_edit": "Keep me"}} + ) + assert publish.publish_project_snapshot(envelope, **args)["_id"] == receipt["_id"] + assert ( + database.concepts.find_one({"concept_id": TASK})["metadata"][ + "later_accepted_edit" + ] + == "Keep me" + ) + assert database.task_dispatch_authorities.find_one({"_id": TASK})["actor_concept_id"] is None + + +def test_changed_destination_is_not_replaced_by_an_older_preflight(scenario): + database, envelope, args = scenario + database.concepts.update_one( + {"concept_id": TASK}, + { + "$push": { + "metadata.comments": {"id": "local", "body": "Accepted local change"} + } + }, + ) + with pytest.raises(ValueError, match="changed after transfer preflight"): + publish.publish_project_snapshot(envelope, **args) + assert database.concepts.find_one({"concept_id": PROJECT}) is None + assert ( + database.concepts.find_one({"concept_id": TASK})["metadata"]["comments"][-1][ + "id" + ] + == "local" + ) + + +def test_text_only_change_after_preflight_is_preserved(scenario): + database, envelope, args = scenario + value_id = database.text_values.insert_one( + {"text": "Later local text", "lang": "en", "fingerprint": "local"} + ).inserted_id + relation_id = database.text_relations.insert_one( + { + "subject_concept_id": TASK, + "predicate": "hasDescription", + "object_text_id": value_id, + } + ).inserted_id + with pytest.raises(ValueError, match="text changed"): + publish.publish_project_snapshot(envelope, **args) + assert database.text_relations.find_one({"_id": relation_id}) + assert database.concepts.find_one({"concept_id": PROJECT}) is None diff --git a/tests/backend/test_task_project_service.py b/tests/backend/test_task_project_service.py index 917d395d..0f06f404 100644 --- a/tests/backend/test_task_project_service.py +++ b/tests/backend/test_task_project_service.py @@ -6,6 +6,11 @@ from src.backend.services import task_project_service as projects +@pytest.fixture(autouse=True) +def no_registered_homes(monkeypatch): + monkeypatch.setattr(projects, "project_home_summaries", lambda ids: {}) + + def test_project_catalogue_does_not_grow_with_retained_source_history(monkeypatch): record = { "name": "Research project", @@ -151,3 +156,32 @@ def test_private_view_does_not_break_shared_project_navigation(monkeypatch): assert projects.list_project_collections( {"collection_concept_ids": ["#V#private", "#V#shared"]} ) == [{"concept_id": "#V#shared"}] + + +def test_assigned_task_home_projection_does_not_expose_private_project(monkeypatch): + import mongomock + + database = mongomock.MongoClient().test + database.concepts.insert_one( + { + "concept_id": "#V#private_project", + "attributes": { + "task_project": {"writer": "von", "description": "Private project body"} + }, + } + ) + monkeypatch.setattr( + projects.ConceptsRepository, "collection", lambda: database.concepts + ) + monkeypatch.setattr( + projects.ConceptsRepository, + "find_one", + lambda *a, **k: {"metadata": {"project_concept_id": "#V#private_project"}}, + ) + assert projects.task_execution_home("#V#assigned_task") == { + "writer": "von", + "write_home": None, + } + monkeypatch.setattr(projects.ConceptsRepository, "find_one", lambda *a, **k: None) + with pytest.raises(ValueError, match="not accessible"): + projects.task_execution_home("#V#unshared_task") diff --git a/tests/backend/test_task_project_transfer_service.py b/tests/backend/test_task_project_transfer_service.py index 1d2b6111..28cde844 100644 --- a/tests/backend/test_task_project_transfer_service.py +++ b/tests/backend/test_task_project_transfer_service.py @@ -1,4 +1,3 @@ -import copy from datetime import UTC, datetime import mongomock @@ -25,6 +24,7 @@ def fixture(monkeypatch, tmp_path): transfer.TextValuesRepository, "collection", lambda: database.text_values ) monkeypatch.setattr(transfer, "can_access_concept", lambda concept_id: True) + monkeypatch.setattr(transfer, "filter_accessible_concept_ids", lambda ids: set(ids)) monkeypatch.setattr(transfer, "get_effective_user_concept_id", lambda: USER) monkeypatch.setattr(transfer, "get_project_home", lambda project_id: None) secret = tmp_path / "key" @@ -130,6 +130,9 @@ def test_same_slug_is_not_an_audience_identity_proof(fixture): def test_inaccessible_member_prevents_partial_export(fixture, monkeypatch): _, source, _ = fixture monkeypatch.setattr(transfer, "can_access_concept", lambda cid: cid != TASK) + monkeypatch.setattr( + transfer, "filter_accessible_concept_ids", lambda ids: set(ids) - {TASK} + ) with pytest.raises(PermissionError, match="inaccessible record"): transfer.capture_project(PROJECT, config=source, recipient="dgx") diff --git a/tests/test_restore_von_db.py b/tests/test_restore_von_db.py index 71558914..d7532a63 100644 --- a/tests/test_restore_von_db.py +++ b/tests/test_restore_von_db.py @@ -37,16 +37,16 @@ def _write_receipt(path: Path, artifact_path: Path, *, db_name: str = "von_db") def _zip_backup(source_root: Path, destination_zip: Path) -> Path: - with zipfile.ZipFile(destination_zip, mode="w", compression=zipfile.ZIP_DEFLATED) as zf: + with zipfile.ZipFile( + destination_zip, mode="w", compression=zipfile.ZIP_DEFLATED + ) as zf: for item in source_root.rglob("*"): if item.is_file(): zf.write(item, item.relative_to(source_root).as_posix()) return destination_zip -def test_dry_run_from_receipt_defaults_to_probe_target( - tmp_path: Path, capsys -) -> None: +def test_dry_run_from_receipt_defaults_to_probe_target(tmp_path: Path, capsys) -> None: backup_root = _make_dump_tree(tmp_path) receipt_path = tmp_path / "launcher_receipt.json" _write_receipt(receipt_path, backup_root) @@ -164,7 +164,7 @@ def _fake_run( assert mongo_uri not in observed["command"] assert "--drop" in observed["command"] assert observed["config_mode"] == 0o600 - assert mongo_uri in observed["config_text"] + assert "mongodb://restore-user:restore-secret@example.invalid/?authSource=von_db" in observed["config_text"] assert not Path(observed["config_path"]).exists() @@ -255,3 +255,22 @@ def restore(command, **kwargs): drop_target=False, ) assert (backup_root / "prelude.json").read_text() == original + + +def test_namespace_restore_removes_database_path_without_changing_authentication(): + from scripts.restore_von_db import _uri_for_namespace_restore + + assert ( + _uri_for_namespace_restore( + "mongodb://fixture:password@localhost:27017/von_db?authSource=admin" + ) + == "mongodb://fixture:password@localhost:27017/?authSource=admin" + ) + assert ( + _uri_for_namespace_restore("mongodb://fixture:password@localhost:27017/von_db") + == "mongodb://fixture:password@localhost:27017/?authSource=von_db" + ) + assert ( + _uri_for_namespace_restore("mongodb://localhost:27017/von_db") + == "mongodb://localhost:27017/" + ) From 2a4fb98ac95d6072abd050f62e6b6ba0f290ac93 Mon Sep 17 00:00:00 2001 From: witbrock Date: Wed, 16 Sep 2026 12:15:25 -0400 Subject: [PATCH 3/3] Preserve deliberate personal task assignment across identity redaction --- .../task_dispatch_authority_service.py | 15 +++++++-- .../services/task_management_service.py | 31 ++++++++++--------- .../test_task_dispatch_authority_service.py | 12 +++++-- tests/backend/test_task_management_service.py | 14 +++++++++ 4 files changed, 53 insertions(+), 19 deletions(-) diff --git a/src/backend/services/task_dispatch_authority_service.py b/src/backend/services/task_dispatch_authority_service.py index 006eef87..4dfaca6a 100644 --- a/src/backend/services/task_dispatch_authority_service.py +++ b/src/backend/services/task_dispatch_authority_service.py @@ -11,7 +11,12 @@ import uuid from ..db.mongo_client import get_db -from ..security.access_control import get_effective_user_concept_id +from ..security.access_control import ( + AUTHENTICATED_SESSION_ACTOR_SOURCE, + AUTHENTICATED_SESSION_DERIVED_ACTOR_SOURCE, + TRUSTED_IN_PROCESS_ACTOR_SOURCE, + get_effective_user_concept_id_with_source, +) COLLECTION = "task_dispatch_authorities" @@ -24,8 +29,12 @@ def _collection(): def record_assignment(task: dict) -> dict | None: - actor = get_effective_user_concept_id() - if not actor: + actor, source = get_effective_user_concept_id_with_source() + if not actor or source not in { + AUTHENTICATED_SESSION_ACTOR_SOURCE, + AUTHENTICATED_SESSION_DERIVED_ACTOR_SOURCE, + TRUSTED_IN_PROCESS_ACTOR_SOURCE, + }: revoke_assignment(task["task_concept_id"]) return None task_id = task["task_concept_id"] diff --git a/src/backend/services/task_management_service.py b/src/backend/services/task_management_service.py index 4b369ef0..615d1240 100644 --- a/src/backend/services/task_management_service.py +++ b/src/backend/services/task_management_service.py @@ -2465,19 +2465,12 @@ def assign_task( existing_assignees = [existing_assignees] if not isinstance(existing_assignees, list): existing_assignees = [] - for old_assignee in existing_assignees: - ConceptsRepository.mutate_relationship_edge( - source_id=task_concept_id, - kind=PREDICATE_HAS_ASSIGNEE, - target_id=old_assignee, - action="remove", - ) - - ConceptsRepository.mutate_relationship_edge( - source_id=task_concept_id, - kind=PREDICATE_HAS_ASSIGNEE, - target_id=assignee_concept_id, - action="add", + # Related identities may be redacted in a personal scope. Replace the + # singleton edge exactly; enumerating the visible old targets can leave an + # invisible previous assignee in place. + ConceptsRepository.update_one( + {"concept_id": task_concept_id}, + {"$set": {f"relationships.{PREDICATE_HAS_ASSIGNEE}": [assignee_concept_id]}}, ) # Source assignment records do not grant the source participant new access. @@ -2511,9 +2504,19 @@ def assign_task( except Exception as e: logger.debug("Failed to append task assignment history event: %s", e) assigned = get_task(task_concept_id) - if update_visibility and assigned.get("assignee_concept_id") == assignee_concept_id: + if update_visibility and ConceptsRepository.find_one( + { + "concept_id": task_concept_id, + f"relationships.{PREDICATE_HAS_ASSIGNEE}": [assignee_concept_id], + }, + projection={"_id": 1}, + ): from .task_dispatch_authority_service import record_assignment + # The actor supplied this identity and the actor-scoped query verified + # its exact edge. Do not require visibility of the agent's profile to + # acknowledge the actor's own assignment or establish dispatch. + assigned["assignee_concept_id"] = assignee_concept_id receipt = record_assignment(assigned) if receipt: ConceptsRepository.update_one( diff --git a/tests/backend/test_task_dispatch_authority_service.py b/tests/backend/test_task_dispatch_authority_service.py index b942d67c..c3de84bb 100644 --- a/tests/backend/test_task_dispatch_authority_service.py +++ b/tests/backend/test_task_dispatch_authority_service.py @@ -30,10 +30,10 @@ def test_transferred_assignment_needs_current_actor_receipt_without_rewriting_cr assert not worker.authorised_task(task, config) task["dispatch_requested"] = True assert not worker.authorised_task(task, config) - monkeypatch.setattr(authority, "get_effective_user_concept_id", lambda: "#V#other") + monkeypatch.setattr(authority, "get_effective_user_concept_id_with_source", lambda: ("#V#other", authority.TRUSTED_IN_PROCESS_ACTOR_SOURCE)) authority.record_assignment(task) assert not worker.authorised_task(task, config) - monkeypatch.setattr(authority, "get_effective_user_concept_id", lambda: "#V#owner") + monkeypatch.setattr(authority, "get_effective_user_concept_id_with_source", lambda: ("#V#owner", authority.TRUSTED_IN_PROCESS_ACTOR_SOURCE)) authority.record_assignment(task) assert worker.authorised_task(task, config) assert task["created_by_concept_id"] == "#V#historical_author" @@ -74,3 +74,11 @@ def test_only_active_local_home_can_dispatch(monkeypatch, state, node, allowed): ) is allowed ) + + +def test_legacy_identity_header_cannot_grant_dispatch(monkeypatch): + collection = mongomock.MongoClient().test.receipts + monkeypatch.setattr(authority, "_collection", lambda: collection) + monkeypatch.setattr(authority, "get_effective_user_concept_id_with_source", lambda: ("#V#owner", "legacy_identity_header")) + assert authority.record_assignment({"task_concept_id": "#V#task"}) is None + assert collection.find_one({"_id": "#V#task"})["revoked"] is True diff --git a/tests/backend/test_task_management_service.py b/tests/backend/test_task_management_service.py index 0f9be66a..64da0ebb 100644 --- a/tests/backend/test_task_management_service.py +++ b/tests/backend/test_task_management_service.py @@ -672,6 +672,20 @@ def test_assign_task_success( mock_repo.mutate_relationship_edge.assert_called() +class TestPersonalTaskAssignment: + @patch("src.backend.services.task_dispatch_authority_service.record_assignment", return_value={"dispatch_id": "fresh"}) + @patch("src.backend.services.task_management_service.get_task") + @patch("src.backend.services.task_management_service.ConceptsRepository") + def test_redacted_agent_profile_does_not_hide_deliberate_assignment(self, repo, get_task_mock, record): + repo.find_one.return_value = {"concept_id": "#V#task_abc", "relationships": {"is_an_instance_of": [TASK_SPECIFICATION_TYPE_ID]}} + get_task_mock.return_value = {"task_concept_id": "#V#task_abc", "assignee_concept_id": None, "organisation_concept_id": None} + result = assign_task("#V#task_abc", "#V#agent") + assert result["assignee_concept_id"] == "#V#agent" + assert result["dispatch_requested"] is True + assert record.call_args.args[0]["assignee_concept_id"] == "#V#agent" + repo.update_one.assert_any_call({"concept_id": "#V#task_abc"}, {"$set": {"relationships.#V#hasAssignee": ["#V#agent"]}}) + + class TestGetTasksForUser: """Tests for get_tasks_for_user function."""