diff --git a/docs/engineering/jvnautosci_2728_article_checkpoint_incident.md b/docs/engineering/jvnautosci_2728_article_checkpoint_incident.md new file mode 100644 index 00000000..dbb799e4 --- /dev/null +++ b/docs/engineering/jvnautosci_2728_article_checkpoint_incident.md @@ -0,0 +1,72 @@ +# Article checkpoint recovery incident, 7 September 2026 + +- **Kind:** Incident evidence record +- **Authority:** Observations and causal interpretation; live repair and delivery decision in [JVNAUTOSCI-2728](https://naoinstitute.atlassian.net/browse/JVNAUTOSCI-2728) +- **Request:** `804d2fe7-4148-4859-9c9b-f94f7829a65e` +- **Workflow instance:** `4881438b-3281-4b0f-aeb4-58a1ebb874f5` +- **Original producer:** `e7a0490cf832df19ccf74e0da82bf3f41dd814dc` + +## Observed chronology + +1. The user requested representation of *An Alien Mind*. The metadata workflow + began at 22:25:59 UTC. An awaited tool call returned a running partial receipt + after its 90-second observation interval; the turn finalised around 22:28:08. + This was an observation expiry, not a transport failure. The model checked + `article_concept_id`, although the workflow contract returns `paper_concept_id`. +2. The article already existed by 22:26:41. Author resolution and canonical + authorship read-back completed by 22:29:19. The saved checkpoint contains + successful article and text-relation reads. The source URL from the original + mail was omitted from the launch inputs and was not stored on the article. +3. At 22:29:42 the complete workflow context was enqueued to the original + checkout's local spillway. Its compressed SHA-256 is + `f4a190ee9a578bc9ed7c2e7ea0acb278b5baaa80473a0980d1f627e8823c1480`. + The checkpoint advanced to `summarise_representation_evidence`. +4. A later restart ran from `Von-runtime-main`. The successor worker claimed the + instance at 22:34:48 and could not find that checkpoint in its own relative + spillway directory. The original bytes remained in `Von/data/blob_spillway`. +5. Execution hydration returned the failed blob reference as if it were usable + workflow context. The summary received no article context, correctly returned + `verification_passed=false`, and the unconditional success transition still + completed the workflow at 22:35:17. + +The restart happened after the original partial response. It explains the +subsequent failed recovery, not the original decision to stop observing. + +## Causal boundaries + +- Local stdio bound operator provenance to selected Gmail/conversation tools, + while telemetry wrappers overwrote provenance with `tool_payload_fallback`. + Thus diagnostics were denied and an existing workflow appeared not found. +- After authority was restored, the original diagnostic and workflow responses + exceeded the stdio response limit. Existing bounded telemetry paging could + serve those records without changing authority or raising the response limit. +- A working-directory-dependent spillway location stranded acknowledged pending + blobs when the runtime moved between linked checkouts. +- Fail-soft loading intended for auxiliary evidence also accepted a missing + top-level execution checkpoint. The saved blob wrapper was truthy, so the + executor resumed the summary state with no domain context. +- The represented summary stage treated successful JSON generation as workflow + success, independently of its own verification result. Its author context also + expected an older aggregate output key instead of the actual author records + and authorship receipts. + +## Canonical artefacts and evidence + +The article is `#V#external_identity_bibliographic_110c3e4b9ddc8590`. Its authorship +relation targets `#V#external_identity_scholarly_author_occurrence_fe858845354ff098`. +Both were read back through canonical tools before recovery. The source URL was +confirmed from the original Gmail message, with tracking parameters removed: +`https://openai.com/index/an-alien-mind/`. + +The original checkpoint and live workflow definition were saved before repair. +Runtime-local blobs were copied to the shared primary spillway with byte/hash +checks, no overwrites, and all originals retained. Raw telemetry and authority +carriers remain in private operational storage, not this repository. + +Targeted evidence covers local stdio authority, signed-reference restrictions, +bounded response reconstruction, cross-worktree pending-blob access, missing +essential context, positive/negative summary verification, and article reuse. +The broader workflow-tools suite was interrupted during an external SSL wait; +its partial run is not claimed as a completed validation campaign. + +See Jira for live recovery receipts, activation identity and publication status. diff --git a/docs/engineering/operational_engineering_guide.md b/docs/engineering/operational_engineering_guide.md index cae4810a..e1ade418 100644 --- a/docs/engineering/operational_engineering_guide.md +++ b/docs/engineering/operational_engineering_guide.md @@ -153,7 +153,18 @@ unchanged verification when needed, and published a current dependency receipt. This command is not a startup workaround and must not be scheduled on every restart. -### 3.4 Clean up repeated local helpers +### 3.4 Keep pending blobs accessible across runtime checkouts + +The default blob spillway is `data/blob_spillway` in the main Git checkout, +shared by linked worktrees. `VON_BLOB_SPILLWAY_DIR` remains an explicit override; +use an absolute path when multiple processes must share pending blobs. Before +switching an existing deployment from a worktree-local queue, copy its pending +blobs and manifests into the shared directory, check immutable blob hashes, +and preserve the originals until recovery is verified. Missing complete inputs +or workflow checkpoints block execution; diagnostic reads retain the failed +reference so the original bytes can be recovered. + +### 3.5 Clean up repeated local helpers Before starting another server, browser replay, or MCP-heavy batch after several retries, inspect for: diff --git a/docs/engineering/security_considerations.md b/docs/engineering/security_considerations.md index e2dfc635..44b95f87 100644 --- a/docs/engineering/security_considerations.md +++ b/docs/engineering/security_considerations.md @@ -359,6 +359,10 @@ configuration cannot silently reuse the previous account's proxy. surfaces and may use operator-supplied scope. They are not part of the ordinary actor-scoped projection and must not be described as though every MCP route shared its identity model. +- Local stdio diagnostic reads use an explicit server-side operator allow-list + in `mcp_stdio_server.py`. Existing actor contexts are preserved; supplied + telemetry references retain their signed target checks. This does not grant + operator provenance to workflow execution, recovery, or ontology mutation. - External MCP servers (arXiv, future integrations) may not respect namespace - No rate limiting on tool invocations - No audit trail of tool access by user diff --git a/src/backend/integrations/internal_mcp/catalogue.py b/src/backend/integrations/internal_mcp/catalogue.py index 8fb41ff9..fedfcb6f 100644 --- a/src/backend/integrations/internal_mcp/catalogue.py +++ b/src/backend/integrations/internal_mcp/catalogue.py @@ -18530,6 +18530,10 @@ def _turn_execution_namespace_coverage_report(**kwargs): "created_at", "identifier_binding", "history_coverage", + "instance_id", + "workflow_id", + "status", + "current_state", ) @@ -18586,6 +18590,20 @@ def _bounded_delegated_telemetry_payload( if not delegated: return dict(payload) + return _bounded_telemetry_payload( + payload, arguments=arguments, artifact_kind=artifact_kind, + preserve_inline_below_limit=preserve_inline_below_limit, + ) + + +def _bounded_telemetry_payload( + payload: Mapping[str, Any], + *, + arguments: Mapping[str, Any], + artifact_kind: str, + preserve_inline_below_limit: bool, +) -> dict[str, Any]: + """Page authorised telemetry independently of its authority carrier.""" import hashlib import json @@ -45733,6 +45751,8 @@ def _build_default_catalogue_task_and_workflow_definitions() -> List[MethodDefin input_schema=Schema( required={"instance_id": str}, optional={ + "offset": int, + "limit": int, "await_terminal": (bool, str, int, float), "advisory_seconds": (int, float), "timeout_seconds": (int, float), @@ -45794,6 +45814,8 @@ def _build_default_catalogue_task_and_workflow_definitions() -> List[MethodDefin input_schema=Schema( required={}, optional={ + "offset": int, + "limit": int, "execution_id": (str, type(None)), "instance_id": (str, type(None)), }, diff --git a/src/backend/mcp_server/mcp_stdio_server.py b/src/backend/mcp_server/mcp_stdio_server.py index b99936f4..00dc111c 100644 --- a/src/backend/mcp_server/mcp_stdio_server.py +++ b/src/backend/mcp_server/mcp_stdio_server.py @@ -500,6 +500,7 @@ def _bind_imports(module_name: str, names: list[str]) -> None: class VonChatRunCapacityUnavailable(RuntimeError): """Raised when all bounded ``von_chat_run`` worker slots are occupied.""" + _TOOL_LIST_CACHE: list[Tool] | None = None _TOOL_LIST_CACHE_PATH = ( Path(project_root) / "data" / "mcp_tool_cache" / "vontology_tools_runtime.json" @@ -1027,6 +1028,30 @@ async def list_tools() -> list[Tool]: } ) +# Only these read handlers inherit the local stdio operator boundary. Keep +# execution, recovery, schedules and ontology mutations on their own authority +# paths; an input flag or actor identifier must never enlarge this set. +_TRUSTED_LOCAL_OPERATOR_DIAGNOSTIC_READ_TOOLS = frozenset( + { + "chat_history_get_debug_entry", + "chat_history_get_segments", + "conversation_inspect_batch", + "conversation_telemetry_get_locator", + "conversation_transcript_page", + "jira_get_comments", + "jira_get_issue", + "jira_search", + "turn_execution_get", + "turn_execution_get_diagnostics", + "turn_execution_get_live_progress", + "turn_execution_list", + "workflow_get_execution_trace", + "workflow_get_instance", + "workflow_list_execution_traces", + "workflow_list_instances", + } +) + _STDIO_GOVERNED_ONTOLOGY_METHODS = { "create_concepts": "create_concepts", "upsert_text_relation": "upsert_text_relation", @@ -1191,9 +1216,7 @@ async def call_tool(name: str, arguments: Any) -> list[TextContent]: # type: ig "effect_status": "not_started", "mutation_outcome": "not_started", "changed": False, - "error_code": ( - "ontology_sessionless_delegation_not_supported" - ), + "error_code": ("ontology_sessionless_delegation_not_supported"), "error": ( "Sessionless ontology delegation execution is " "unavailable; use an actor-bound trusted surface." @@ -1201,6 +1224,42 @@ async def call_tool(name: str, arguments: Any) -> list[TextContent]: # type: ig } ) ] + if name in _TRUSTED_LOCAL_OPERATOR_DIAGNOSTIC_READ_TOOLS: + from src.backend.security.access_control import ( + get_effective_organisation_concept_id, + get_effective_user_concept_id, + ) + + # A supplied delegation keeps its narrower target and validation; + # never silently replace an existing actor context with operator. + source = internal_mcp_gateway_module.get_internal_mcp_actor_context_source() + actor = ( + internal_mcp_gateway_module.get_internal_mcp_preexisting_actor_context() + ) + actor = actor or ( + get_effective_user_concept_id(), + get_effective_organisation_concept_id(), + ) + has_reference = any( + key in parsed_arguments + for key in ( + "conversation_ref", + "history_location_ref", + "turn_telemetry_ref", + ) + ) + if has_reference: + source = "tool_payload_fallback" + elif source is None: + source = ( + "preexisting_authenticated_or_workflow_context" + if any(actor) + else internal_mcp_gateway_module.INTERNAL_MCP_TRUSTED_LOCAL_OPERATOR_SOURCE + ) + with bind_internal_mcp_actor_context_source( + source, preexisting_actor_context=actor if any(actor) else None + ): + return await handler(parsed_arguments) if name in ( _TRUSTED_LOCAL_OPERATOR_GMAIL_TOOLS | _TRUSTED_LOCAL_OPERATOR_CONVERSATION_TOOLS @@ -3430,6 +3489,39 @@ def _run_catalogue_proxy_handler( ] +def _run_diagnostic_read_proxy_handler( + handler: Callable[..., Any], + arguments: dict[str, Any], + **kwargs: Any, +) -> list[TextContent]: + """Preserve entry-point authority; a direct handler call cannot mint it.""" + + source = internal_mcp_gateway_module.get_internal_mcp_actor_context_source() + actor = internal_mcp_gateway_module.get_internal_mcp_preexisting_actor_context() + with bind_internal_mcp_actor_context_source( + source or "tool_payload_fallback", preexisting_actor_context=actor + ): + # Reuse the signed-reference transport pager for local operator reads. + # Do not page errors or re-page an already bounded delegated response. + def bounded_handler(**payload_args: Any) -> Any: + result = handler(**payload_args) + if ( + isinstance(result, dict) + and result.get("success") is not False + and "bounded_read" not in result + and len(json.dumps(result, separators=(",", ":"), default=str)) + > _get_stdio_max_response_chars() + ): + return internal_mcp_catalogue_module._bounded_telemetry_payload( + result, arguments=arguments, + artifact_kind=handler.__name__.lstrip("_"), + preserve_inline_below_limit=True, + ) + return result + + return _run_catalogue_proxy_handler(bounded_handler, arguments, **kwargs) + + def _run_untrusted_workflow_proxy_handler( handler: Callable[..., Any], arguments: dict[str, Any], @@ -3445,7 +3537,7 @@ def _run_untrusted_workflow_proxy_handler( async def _handle_jira_search(arguments: dict[str, Any]) -> list[TextContent]: - return _run_catalogue_proxy_handler( + return _run_diagnostic_read_proxy_handler( _jira_search, arguments, tool_family_label="Jira", @@ -3454,7 +3546,7 @@ async def _handle_jira_search(arguments: dict[str, Any]) -> list[TextContent]: async def _handle_jira_get_comments(arguments: dict[str, Any]) -> list[TextContent]: - return _run_catalogue_proxy_handler( + return _run_diagnostic_read_proxy_handler( _jira_get_comments, arguments, tool_family_label="Jira", @@ -3463,7 +3555,7 @@ async def _handle_jira_get_comments(arguments: dict[str, Any]) -> list[TextConte async def _handle_jira_get_issue(arguments: dict[str, Any]) -> list[TextContent]: - return _run_catalogue_proxy_handler( + return _run_diagnostic_read_proxy_handler( _jira_get_issue, arguments, tool_family_label="Jira", @@ -3755,18 +3847,20 @@ async def _handle_workflow_execute(arguments: dict[str, Any]) -> list[TextConten async def _handle_workflow_list_instances( arguments: dict[str, Any], ) -> list[TextContent]: - return _run_untrusted_workflow_proxy_handler( + return _run_diagnostic_read_proxy_handler( _workflow_list_instances, arguments, + tool_family_label="Diagnostics", ) async def _handle_workflow_list_execution_traces( arguments: dict[str, Any], ) -> list[TextContent]: - return _run_untrusted_workflow_proxy_handler( + return _run_diagnostic_read_proxy_handler( _workflow_list_execution_traces, arguments, + tool_family_label="Diagnostics", ) @@ -3780,18 +3874,20 @@ async def _handle_workflow_build_prediction_envelope( async def _handle_workflow_get_instance(arguments: dict[str, Any]) -> list[TextContent]: - return _run_untrusted_workflow_proxy_handler( + return _run_diagnostic_read_proxy_handler( _workflow_get_instance, arguments, + tool_family_label="Diagnostics", ) async def _handle_workflow_get_execution_trace( arguments: dict[str, Any], ) -> list[TextContent]: - return _run_untrusted_workflow_proxy_handler( + return _run_diagnostic_read_proxy_handler( _workflow_get_execution_trace, arguments, + tool_family_label="Diagnostics", ) @@ -3877,18 +3973,20 @@ async def _handle_workflow_trigger_schedule( async def _handle_chat_history_get_segments( arguments: dict[str, Any], ) -> list[TextContent]: - return _run_untrusted_workflow_proxy_handler( + return _run_diagnostic_read_proxy_handler( _chat_history_get_segments, arguments, + tool_family_label="Diagnostics", ) async def _handle_chat_history_get_debug_entry( arguments: dict[str, Any], ) -> list[TextContent]: - return _run_untrusted_workflow_proxy_handler( + return _run_diagnostic_read_proxy_handler( _chat_history_get_debug_entry, arguments, + tool_family_label="Diagnostics", ) @@ -3915,7 +4013,7 @@ async def _handle_conversation_get( async def _handle_conversation_transcript_page( arguments: dict[str, Any], ) -> list[TextContent]: - return _run_catalogue_proxy_handler( + return _run_diagnostic_read_proxy_handler( _conversation_transcript_page, arguments, tool_family_label="Conversation", @@ -3925,7 +4023,7 @@ async def _handle_conversation_transcript_page( async def _handle_conversation_inspect_batch( arguments: dict[str, Any], ) -> list[TextContent]: - return _run_catalogue_proxy_handler( + return _run_diagnostic_read_proxy_handler( _conversation_inspect_batch, arguments, tool_family_label="Conversation", @@ -3965,32 +4063,36 @@ async def _handle_conversation_manage_batch( async def _handle_conversation_telemetry_get_locator( arguments: dict[str, Any], ) -> list[TextContent]: - return _run_untrusted_workflow_proxy_handler( + return _run_diagnostic_read_proxy_handler( _conversation_telemetry_get_locator, arguments, + tool_family_label="Diagnostics", ) async def _handle_turn_execution_list(arguments: dict[str, Any]) -> list[TextContent]: - return _run_untrusted_workflow_proxy_handler( + return _run_diagnostic_read_proxy_handler( _turn_execution_list, arguments, + tool_family_label="Diagnostics", ) async def _handle_turn_execution_get(arguments: dict[str, Any]) -> list[TextContent]: - return _run_untrusted_workflow_proxy_handler( + return _run_diagnostic_read_proxy_handler( _turn_execution_get, arguments, + tool_family_label="Diagnostics", ) async def _handle_turn_execution_get_diagnostics( arguments: dict[str, Any], ) -> list[TextContent]: - return _run_untrusted_workflow_proxy_handler( + return _run_diagnostic_read_proxy_handler( _turn_execution_get_diagnostics, arguments, + tool_family_label="Diagnostics", ) @@ -4017,9 +4119,10 @@ async def _handle_failure_case_reference_resolve( async def _handle_turn_execution_get_live_progress( arguments: dict[str, Any], ) -> list[TextContent]: - return _run_untrusted_workflow_proxy_handler( + return _run_diagnostic_read_proxy_handler( _turn_execution_get_live_progress, arguments, + tool_family_label="Diagnostics", ) diff --git a/src/backend/mcp_server/vontology_mcp.json b/src/backend/mcp_server/vontology_mcp.json index da5c0f51..788d7afd 100644 --- a/src/backend/mcp_server/vontology_mcp.json +++ b/src/backend/mcp_server/vontology_mcp.json @@ -7895,6 +7895,12 @@ "inputSchema": { "type": "object", "properties": { + "offset": { + "type": "integer" + }, + "limit": { + "type": "integer" + }, "execution_id": { "type": [ "string", @@ -7922,6 +7928,12 @@ "instance_id": { "type": "string" }, + "offset": { + "type": "integer" + }, + "limit": { + "type": "integer" + }, "await_terminal": { "type": [ "boolean", diff --git a/src/backend/services/blob_spillway.py b/src/backend/services/blob_spillway.py index dffe215b..2e698d09 100644 --- a/src/backend/services/blob_spillway.py +++ b/src/backend/services/blob_spillway.py @@ -12,7 +12,7 @@ Env vars: VON_BLOB_SPILLWAY_ENABLED — "true" (default) / "false". - VON_BLOB_SPILLWAY_DIR — local root dir (default: data/blob_spillway). + VON_BLOB_SPILLWAY_DIR — local root dir (default: main checkout data/blob_spillway). VON_BLOB_SPILLWAY_MAX_RETRIES — int (default: 10). VON_BLOB_SPILLWAY_MIGRATE_INTERVAL_SECONDS — float (default: 30, used by caller). @@ -404,7 +404,30 @@ def cleanup_committed_cache( def _default_spillway_dir() -> Path: - return Path(os.getenv("VON_BLOB_SPILLWAY_DIR", _DEFAULT_SPILLWAY_DIR)) + configured = os.getenv("VON_BLOB_SPILLWAY_DIR") + if configured: + return Path(configured) + + # Linked runtime worktrees share pending checkpoints with their main + # checkout. A relative process working directory must not decide whether + # an already acknowledged blob can be recovered after deployment. + repo_root = Path(__file__).resolve().parents[3] + git_dir = repo_root / ".git" + try: + if git_dir.is_file(): + marker = git_dir.read_text(encoding="utf-8").strip() + if not marker.startswith("gitdir: "): + raise ValueError("Invalid worktree gitdir marker") + git_dir = (repo_root / marker.removeprefix("gitdir: ")).resolve() + common_file = git_dir / "commondir" + if common_file.is_file(): + git_dir = (git_dir / common_file.read_text(encoding="utf-8").strip()).resolve() + if git_dir.is_dir() and git_dir.name == ".git": + return git_dir.parent / _DEFAULT_SPILLWAY_DIR + except (OSError, ValueError) as exc: + logger.warning("Cannot resolve shared spillway checkout: %s", exc) + # Non-Git installations retain their configured working-directory layout. + return Path(_DEFAULT_SPILLWAY_DIR) def _default_max_retries() -> int: diff --git a/src/backend/workflows/durable/instance_manager.py b/src/backend/workflows/durable/instance_manager.py index 24fb3370..d33cfc65 100644 --- a/src/backend/workflows/durable/instance_manager.py +++ b/src/backend/workflows/durable/instance_manager.py @@ -642,12 +642,18 @@ def _has_unhydrated_structured_continuation(candidate: Any) -> bool: continue hydrated = hydrate_workflow_payload_blob_refs( hydrated_doc[field], - # Ordinary diagnostic/auxiliary blobs remain fail-soft even for - # execution. Only provider continuation state is essential to - # avoid repeating tools or inventing an uncorrelated answer. + # Diagnostic/auxiliary blobs may remain fail-soft. The complete + # inputs/checkpoint and provider continuation are execution + # state, not optional evidence; validate them below. fail_soft=True, ) hydrated_doc[field] = hydrated.payload + if ( + not fail_soft + and field in {"inputs", "workflow_data"} + and is_workflow_payload_blob_ref(hydrated.payload) + ): + raise RuntimeError(f"workflow_{field}_hydration_failed") if not fail_soft and _has_unhydrated_structured_continuation( hydrated.payload ): @@ -1100,9 +1106,9 @@ def get_instance( Args: instance_id: The instance identifier. - for_execution: Fail closed if offloaded structured-tool - continuation state cannot be hydrated. Other auxiliary blob - failures remain visible but fail-soft. + for_execution: Fail closed if complete inputs/checkpoint or + structured-tool continuation cannot be hydrated. Auxiliary + diagnostic blob failures remain visible but fail-soft. Returns: WorkflowInstance if found, None otherwise. diff --git a/src/backend/workflows/repo_seed_bundles/paper_representation_workflow_seed_bundle.json b/src/backend/workflows/repo_seed_bundles/paper_representation_workflow_seed_bundle.json index 74cc814c..e690bd59 100644 --- a/src/backend/workflows/repo_seed_bundles/paper_representation_workflow_seed_bundle.json +++ b/src/backend/workflows/repo_seed_bundles/paper_representation_workflow_seed_bundle.json @@ -2,7 +2,7 @@ "family_id": "paper_representation_workflow_seed_bundle", "managed_by": "paper_representation_workflow_vontology_service", "schema_version": "repo_seed_workflow_bundle.v1", - "seed_version": "37", + "seed_version": "39", "known_legacy_authority_payload_sha256_by_seed_version": { "#V#arxiv_paper_representation_workflow": { "18": [ @@ -51,7 +51,7 @@ ] } }, - "source_tag": "JVNAUTOSCI-2244", + "source_tag": "JVNAUTOSCI-2728", "supported_action_ids": [ "scholarly_paper.normalise_inputs", "scholarly_paper.normalise_external_identity", @@ -90,8 +90,8 @@ { "workflow_id": "#V#scholarly_article_metadata_representation_workflow", "display_name": "Scholarly Article Metadata Representation Workflow", - "description": "Canonical compositional workflow for representing non-arXiv scholarly article metadata, DOI/source URLs, ACM/DOI pages, and pasted bibliographic metadata without requiring a source-specific PDF, arXiv identifier, or file copy.", - "content": "Represent a scholarly article from a DOI URL, source URL, pasted bibliographic text, or normalised metadata by extracting/coalescing metadata fields, creating or reusing a #V#scholarly_article concept through generic Vontology write tools, attaching text metadata, resolving authors/topics, attaching DOI and source identifiers, optionally linking a file copy, and reading back the materialised concept. Sparse URL-only inputs may create a provisional article representation named by DOI/source URI while preserving provenance. This workflow is the preferred execution surface for non-arXiv article representation requests before falling back to the generic tool-calling write path.", + "description": "Canonical compositional workflow for representing non-arXiv scholarly article metadata, DOI/source URLs, ACM/DOI pages, and pasted bibliographic metadata without requiring a source-specific PDF, arXiv identifier, or file copy. Preserve the supplied source URL as source_uri. Observe a running partial result on the same instance until terminal evidence is available; paper_concept_id is the article output key.", + "content": "Represent a scholarly article from a DOI URL, source URL, pasted bibliographic text, or normalised metadata by extracting/coalescing metadata fields, creating or reusing a #V#scholarly_article concept through generic Vontology write tools, attaching text metadata, resolving authors/topics, attaching DOI and source identifiers, optionally linking a file copy, and reading back the materialised concept. Sparse URL-only inputs may create a provisional article representation named by DOI/source URI while preserving provenance. This workflow is the preferred execution surface for non-arXiv article representation requests before falling back to the generic tool-calling write path. Carry any already retrieved source URL into source_uri. A partial running receipt is an observation, not completion: continue observing the same instance with workflow_get_instance(await_terminal=true) while progress remains recoverable. Read paper_concept_id from its declared outputs and verify canonical article, authorship and source metadata before reporting completion.", "text_relations": [ { "predicate": "#V#hasWorkflowDiscoveryExemplarsJson", @@ -1885,6 +1885,10 @@ "context_key": "author_concept_ids", "label": "Resolved author concept IDs" }, + { + "context_key": "public_author_records", + "label": "Resolved source-bounded public author records" + }, { "context_key": "doi", "label": "DOI" @@ -1914,7 +1918,7 @@ "validation_policy": { "output_format": "json_value" }, - "next_state": "completed", + "next_state": "failed", "on_failure_state": "failed", "tool_output_mapping_specs": [ { @@ -1943,6 +1947,17 @@ "observations", "metadata_verification", "verification_passed" + ], + "conditional_transitions": [ + { + "to_state": "completed", + "reason": "representation_evidence_verified", + "condition_spec": { + "kind": "context_value_equals", + "key": "verification_passed", + "value": true + } + } ] }, { diff --git a/tests/backend/test_blob_spillway.py b/tests/backend/test_blob_spillway.py index b5ffa005..c2801d0c 100644 --- a/tests/backend/test_blob_spillway.py +++ b/tests/backend/test_blob_spillway.py @@ -646,3 +646,27 @@ def test_compact_falls_back_to_remote_when_spillway_disabled(monkeypatch) -> Non blob_ref = messages_ref.get("blob_ref") assert isinstance(blob_ref, dict) assert blob_ref.get("backend") == "s3" + + +def test_linked_worktree_reads_primary_pending_checkpoint(tmp_path, monkeypatch): + from src.backend.services import blob_spillway + + primary = tmp_path / "Von" + runtime = tmp_path / "Von-runtime" + git_dir = primary / ".git" / "worktrees" / "runtime" + git_dir.mkdir(parents=True) + (git_dir / "commondir").write_text("../..\n") + runtime.mkdir() + (runtime / ".git").write_text(f"gitdir: {git_dir}\n") + monkeypatch.delenv("VON_BLOB_SPILLWAY_DIR", raising=False) + monkeypatch.setattr(blob_spillway, "__file__", str(primary / "src/backend/services/blob_spillway.py")) + original_root = blob_spillway._default_spillway_dir() + writer = BlobSpillwayQueue(original_root) + writer.enqueue(key="workflow/checkpoint.json", data=b'{"paper_concept_id":"paper"}') + monkeypatch.chdir(runtime) + monkeypatch.setattr(blob_spillway, "__file__", str(runtime / "src/backend/services/blob_spillway.py")) + recovered_root = blob_spillway._default_spillway_dir() + assert recovered_root == original_root + assert BlobSpillwayQueue(recovered_root).get_local_bytes("workflow/checkpoint.json") == b'{"paper_concept_id":"paper"}' + monkeypatch.setenv("VON_BLOB_SPILLWAY_DIR", str(tmp_path / "isolated")) + assert blob_spillway._default_spillway_dir() == tmp_path / "isolated" diff --git a/tests/backend/test_jira_read_authority.py b/tests/backend/test_jira_read_authority.py index ccfb5c8f..af0cb6b7 100644 --- a/tests/backend/test_jira_read_authority.py +++ b/tests/backend/test_jira_read_authority.py @@ -258,3 +258,14 @@ def run(_options): assert calls == [True] assert not internal_mcp_actor_context_is_trusted_local_operator() assert '"dry_run": true' in capsys.readouterr().out + + +@pytest.mark.parametrize("name,arguments", READS) +def test_local_stdio_jira_reads_reach_account_rpc(account, name, arguments): + import json + from src.backend.mcp_server import mcp_stdio_server + + blocks = asyncio.run(mcp_stdio_server.call_tool(name, arguments)) + result = json.loads(blocks[0].text) + assert result.get("success") is not False + assert account["calls"] == [(name, arguments)] diff --git a/tests/backend/test_mcp_stdio_diagnostic_authority.py b/tests/backend/test_mcp_stdio_diagnostic_authority.py new file mode 100644 index 00000000..ee453ffe --- /dev/null +++ b/tests/backend/test_mcp_stdio_diagnostic_authority.py @@ -0,0 +1,158 @@ +"""Local stdio diagnostic authority reaches storage without escaping its call.""" + +import asyncio +import json + +import pytest + +from src.backend.integrations.internal_mcp import catalogue, gateway +from src.backend.mcp_server import mcp_stdio_server as stdio +from src.backend.security.access_control import override_current_actor + + +def decode(blocks): + return json.loads(blocks[0].text) + + +@pytest.fixture +def diagnostics(monkeypatch): + calls = [] + + def read(**kwargs): + calls.append(kwargs) + return {"success": True, "request_id": kwargs["request_id"]} + + monkeypatch.setattr( + "src.backend.services.turn_execution_diagnostics_service.get_turn_execution_diagnostics_payload", + read, + ) + return calls + + +def test_local_stdio_reads_exact_request_and_does_not_leak_authority(diagnostics): + async def run(): + result = decode( + await stdio.call_tool( + "turn_execution_get_diagnostics", {"request_id": "incident-turn"} + ) + ) + assert gateway.get_internal_mcp_actor_context_source() is None + # The same handler outside the local transport still fails closed. + denied = decode( + await stdio._handle_turn_execution_get_diagnostics( + { + "request_id": "other-turn", + "operator": True, + "user_concept_id": "#V#admin", + } + ) + ) + return result, denied + + result, denied = asyncio.run(run()) + assert result["request_id"] == "incident-turn" + assert result["success"] is True + assert denied["error_code"] == "workflow_global_admin_authority_required" + assert [call["request_id"] for call in diagnostics] == ["incident-turn"] + + +def test_stdio_cannot_upgrade_existing_authenticated_actor(monkeypatch, diagnostics): + monkeypatch.setattr( + "src.backend.services.von_operational_administrator_service.is_live_von_operational_administrator", + lambda _actor: False, + ) + with override_current_actor("#V#ordinary_user", "#V#org"): + result = decode( + asyncio.run( + stdio.call_tool( + "turn_execution_get_diagnostics", {"request_id": "foreign-turn"} + ) + ) + ) + assert result["error_code"] == "workflow_global_admin_authority_required" + assert diagnostics == [] + + +def test_untrusted_gateway_cannot_request_local_operator_authority(diagnostics): + with gateway.bind_internal_mcp_actor_context_source("tool_payload_fallback"): + result = catalogue._turn_execution_get_diagnostics( + request_id="foreign-turn", + operator=True, + user_concept_id="#V#admin", + namespace="admin", + ) + assert result["error_code"] == "workflow_global_admin_authority_required" + assert diagnostics == [] + + +@pytest.mark.parametrize( + "name", + [ + "workflow_execute", + "workflow_cancel_instance", + "workflow_resume_instance", + "workflow_retry_instance", + "workflow_trigger_schedule", + ], +) +def test_diagnostic_read_does_not_authorise_next_workflow_mutation( + monkeypatch, diagnostics, name +): + observed = [] + monkeypatch.setattr( + stdio, "_evaluate_stdio_write_access", lambda *_: (True, {}, {}) + ) + + def mutation(**kwargs): + observed.append(gateway.get_internal_mcp_actor_context_source()) + return {"success": False, "changed": False} + + monkeypatch.setattr(stdio, "_" + name, mutation) + + async def run(): + await stdio.call_tool( + "turn_execution_get_diagnostics", {"request_id": "incident-turn"} + ) + return decode( + await stdio.call_tool(name, {"instance_id": "instance", "operator": True}) + ) + + result = asyncio.run(run()) + assert observed == ["tool_payload_fallback"] + assert result["changed"] is False + + +def test_large_local_diagnostic_can_be_reconstructed_from_bounded_pages(monkeypatch): + import hashlib + + expected = { + "success": True, + "instance_id": "incident", + "workflow_data": {"evidence": '\\"' * 100_000}, + } + monkeypatch.setattr(stdio, "_workflow_get_instance", lambda **_: expected) + monkeypatch.setenv("VON_MCP_STDIO_MAX_RESPONSE_CHARS", "10000") + + async def run(): + chunks = [] + offset = 0 + digest = None + while True: + blocks = await stdio.call_tool( + "workflow_get_instance", + {"instance_id": "incident", "offset": offset, "limit": 20000}, + ) + assert len(blocks[0].text) <= 10000 + page = decode(blocks)["bounded_read"] + assert digest is None or digest == page["sha256"] + digest = page["sha256"] + chunks.append(page["json_chunk"]) + if not page["has_more"]: + break + assert page["next_offset"] > offset + offset = page["next_offset"] + return "".join(chunks), digest + + text, digest = asyncio.run(run()) + assert json.loads(text) == expected + assert hashlib.sha256(text.encode()).hexdigest() == digest diff --git a/tests/backend/test_paper_representation_workflow_vontology_service.py b/tests/backend/test_paper_representation_workflow_vontology_service.py index 0ee8e4fa..954528c5 100644 --- a/tests/backend/test_paper_representation_workflow_vontology_service.py +++ b/tests/backend/test_paper_representation_workflow_vontology_service.py @@ -2528,7 +2528,7 @@ def test_bootstrap_migrates_reviewed_v35_outcome_prompt_map_gap( limit=2, ) assert any( - json.loads(row.get("text") or "{}").get("seed_version") == "37" + json.loads(row.get("text") or "{}").get("seed_version") == "39" for row in marker_rows if isinstance(row.get("text"), str) ) @@ -2658,7 +2658,7 @@ def test_bootstrap_seed_version_refresh_repairs_old_arxiv_launch_contract( for row in marker_rows if isinstance(row.get("text"), str) ] - assert any(payload.get("seed_version") == "37" for payload in marker_payloads) + assert any(payload.get("seed_version") == "39" for payload in marker_payloads) refreshed_definition = load_workflow_definition_from_vontology( ARXIV_PAPER_REPRESENTATION_WORKFLOW_ID @@ -4578,3 +4578,41 @@ def test_live_arxiv_paper_representation_workflow_acceptance_batch( "case_count": len(reports), "failures": failures, } + + +@pytest.mark.parametrize("verified", [True, False, None]) +def test_article_summary_cannot_complete_without_positive_verification( + _canonical_seeded_mock_db: Any, verified +): + from dataclasses import replace + + definition = load_workflow_definition_from_vontology( + SCHOLARLY_ARTICLE_METADATA_REPRESENTATION_WORKFLOW_ID + ) + assert definition is not None + summary_state = next(key for key in definition.states if key.endswith("_summarise_representation_evidence")) + definition = replace(definition, initial_state=summary_state) + + class SummaryLLM: + def generate(self, prompt, context=None, model=None, llm_params=None): + payload = {"response_text": "Read-back result", "observations": [], "metadata_verification": {}, "reasoning": "Observed evidence"} + if verified is not None: + payload["verification_passed"] = verified + return json.dumps(payload) + + result = WorkflowExecutor( + registry=registry_factory.build_durable_action_registry(), max_transitions=3 + ).run( + definition, + environment=WorkflowEnvironment( + llm_client=SummaryLLM(), user_namespace=_LIVE_ARXIV_ACCEPTANCE_NAMESPACE, + user_concept_id=_LIVE_ARXIV_ACCEPTANCE_USER_ID, + org_concept_id=_LIVE_ARXIV_ACCEPTANCE_ORG_ID, + ), + data={"paper_concept_id": "#V#existing_article", "article_readback": {"success": True}}, + ) + assert result.completed is (verified is True) + if verified is None: + assert "verification_passed" in str(result.error) + else: + assert result.final_state.endswith("_completed" if verified else "_failed") diff --git a/tests/backend/test_telemetry_read_delegation_stdio.py b/tests/backend/test_telemetry_read_delegation_stdio.py index 12742703..0e39384c 100644 --- a/tests/backend/test_telemetry_read_delegation_stdio.py +++ b/tests/backend/test_telemetry_read_delegation_stdio.py @@ -291,9 +291,8 @@ def build_locator(**kwargs): ] canonical_json = "".join(page["json_chunk"] for page in page_metadata) assert len(canonical_json) == page_metadata[0]["total_chars"] - assert ( - hashlib.sha256(canonical_json.encode("utf-8")).hexdigest() - == (page_metadata[0]["sha256"]) + assert hashlib.sha256(canonical_json.encode("utf-8")).hexdigest() == ( + page_metadata[0]["sha256"] ) reconstructed_carrier = json.loads(canonical_json) assert reconstructed_carrier["conversation_situation"] == situation @@ -421,9 +420,8 @@ def test_escape_heavy_carrier_pages_remain_under_real_stdio_guard( assert len({page["total_chars"] for page in page_metadata}) == 1 canonical_json = "".join(page["json_chunk"] for page in page_metadata) assert len(canonical_json) == page_metadata[0]["total_chars"] - assert ( - hashlib.sha256(canonical_json.encode("utf-8")).hexdigest() - == (page_metadata[0]["sha256"]) + assert hashlib.sha256(canonical_json.encode("utf-8")).hexdigest() == ( + page_metadata[0]["sha256"] ) reconstructed = json.loads(canonical_json) assert reconstructed["segments"][0][0]["content"] == escape_heavy_content @@ -438,7 +436,7 @@ def test_escape_heavy_carrier_pages_remain_under_real_stdio_guard( assert page["read_delegation"]["delegated_actor_user_id"] == "#V#actor" -def test_stdio_raw_ids_retain_original_denials_and_actor_b_fails_closed() -> None: +def test_untrusted_context_raw_ids_and_actor_b_fail_closed() -> None: actor_a_ref = build_turn_telemetry_binding( request_id="req-2609-a", delegated_actor_user_id="#V#actor_a", @@ -448,23 +446,29 @@ def test_stdio_raw_ids_retain_original_denials_and_actor_b_fails_closed() -> Non organisation_concept_id="#V#org", ) - raw_debug = _stdio_payload( - "chat_history_get_debug_entry", - { - "session_id": "chat-guessed", - "history_index": 2, - "namespace": "#V#actor_a@org", - "user_concept_id": "#V#actor_a", - }, - ) - raw_diagnostics = _stdio_payload( - "turn_execution_get_diagnostics", - { - "request_id": "req-guessed", - "namespace": "#V#actor_a@org", - "user_concept_id": "#V#actor_a", - }, + from src.backend.integrations.internal_mcp.gateway import ( + bind_internal_mcp_actor_context_source, ) + + # Ordinary internal callers do not inherit the standalone local operator. + with bind_internal_mcp_actor_context_source("tool_payload_fallback"): + raw_debug = _stdio_payload( + "chat_history_get_debug_entry", + { + "session_id": "chat-guessed", + "history_index": 2, + "namespace": "#V#actor_a@org", + "user_concept_id": "#V#actor_a", + }, + ) + raw_diagnostics = _stdio_payload( + "turn_execution_get_diagnostics", + { + "request_id": "req-guessed", + "namespace": "#V#actor_a@org", + "user_concept_id": "#V#actor_a", + }, + ) actor_b = _stdio_payload( "turn_execution_get_diagnostics", { @@ -477,13 +481,13 @@ def test_stdio_raw_ids_retain_original_denials_and_actor_b_fails_closed() -> Non ) assert raw_debug["error_code"] == "authenticated_actor_context_required" - assert raw_diagnostics["error_code"] == ( - "workflow_global_admin_authority_required" - ) + assert raw_diagnostics["error_code"] == ("workflow_global_admin_authority_required") assert actor_b["error_code"] == "READ_DELEGATION_ACTOR_MISMATCH" -def test_stdio_delegation_tamper_expiry_tool_and_target_mismatches_fail_closed() -> None: +def test_stdio_delegation_tamper_expiry_tool_and_target_mismatches_fail_closed() -> ( + None +): issued_at = datetime.now(timezone.utc) - timedelta(minutes=5) expired_ref = build_turn_telemetry_binding( request_id="req-expired-2609", @@ -667,9 +671,7 @@ def get_live_progress(**kwargs): }, ) - assert mismatched["error_code"] == ( - "READ_DELEGATION_CANONICAL_TARGET_MISMATCH" - ) + assert mismatched["error_code"] == ("READ_DELEGATION_CANONICAL_TARGET_MISMATCH") def test_stdio_diagnostics_delegation_reads_owner_namespace_for_invitee( diff --git a/tests/backend/test_workflow_payload_blob_offload.py b/tests/backend/test_workflow_payload_blob_offload.py index 8734e62d..d7a0989c 100644 --- a/tests/backend/test_workflow_payload_blob_offload.py +++ b/tests/backend/test_workflow_payload_blob_offload.py @@ -750,3 +750,30 @@ def __getitem__(self, name: str) -> _Collection: loaded = trace_store.get_workflow_execution_trace("execution-1") assert loaded is not None assert loaded["actions"] == actions + + +@pytest.mark.parametrize("field", ["inputs", "workflow_data"]) +def test_missing_complete_execution_payload_remains_recoverable_but_cannot_run(monkeypatch, field): + import copy + + store = _FakeBlobStore() + monkeypatch.setattr("src.backend.services.blob_store.get_blob_store_from_env", lambda: store) + monkeypatch.setenv("VON_WORKFLOW_PAYLOAD_BLOB_THRESHOLD_BYTES", "512") + payload = {"paper_concept_id": "#V#existing_paper", "context": "x" * 300_000} + ref = WorkflowInstanceManager._compact_instance_payload_field( + payload, field=field, instance_id="checkpoint-incident" + ) + assert ref["schema_version"] == "workflow_payload_blob_ref.v1" + doc = {field: ref, "current_state": "summarise_representation_evidence"} + persisted = copy.deepcopy(doc) + saved_bytes = list(store.writes) + store.writes.clear() + degraded = WorkflowInstanceManager._hydrate_instance_payloads(doc) + assert degraded[field]["hydration_error"] + with pytest.raises(RuntimeError, match="hydration_failed"): + WorkflowInstanceManager._hydrate_instance_payloads(doc, fail_soft=False) + assert doc == persisted + store.writes.extend(saved_bytes) + restored = WorkflowInstanceManager._hydrate_instance_payloads(doc, fail_soft=False) + assert restored[field] == payload + assert restored["current_state"] == persisted["current_state"]