diff --git a/Makefile b/Makefile index cdbe6ed..d325eb6 100644 --- a/Makefile +++ b/Makefile @@ -1,10 +1,10 @@ .PHONY: format lint typecheck test verify benchmark format: - python3 -m ruff format src/nexus_runtime tests/test_agent.py tests/test_end_to_end.py tests/test_persistence_replay.py tests/test_scheduler.py tests/test_tools.py tests/test_distributed_runtime.py + python3 -m ruff format src/nexus_runtime tests/test_agent.py tests/test_end_to_end.py tests/test_persistence_replay.py tests/test_scheduler.py tests/test_tools.py tests/test_distributed_runtime.py tests/test_investigation_intelligence.py tests/test_evidence_loop.py tests/test_investigation_application.py tests/test_investigation_cli.py lint: - python3 -m ruff check src/nexus_runtime tests/test_agent.py tests/test_end_to_end.py tests/test_persistence_replay.py tests/test_scheduler.py tests/test_tools.py tests/test_distributed_runtime.py + python3 -m ruff check src/nexus_runtime tests/test_agent.py tests/test_end_to_end.py tests/test_persistence_replay.py tests/test_scheduler.py tests/test_tools.py tests/test_distributed_runtime.py tests/test_investigation_intelligence.py tests/test_evidence_loop.py tests/test_investigation_application.py tests/test_investigation_cli.py typecheck: python3 -m mypy src/nexus_runtime @@ -18,3 +18,4 @@ verify: benchmark: python3 -m nexus_runtime.distributed.benchmark + python3 -m nexus_runtime.investigation.benchmark diff --git a/README.md b/README.md index fbf6b5d..50a2a33 100644 --- a/README.md +++ b/README.md @@ -93,6 +93,25 @@ The Phase 3 milestone provides a provider-neutral distributed execution layer: The public service boundary is intentionally `KnowledgeService`; NEXUS does not read the knowledge engine's database. +### Autonomous investigation (Phase 4) + +Phase 4 closes the epistemic loop: structured objectives are observed against current +knowledge, measurable gaps become scored and budget-aware investigation plans, those +plans execute through the distributed Agent Harness, and provenance-complete evidence +is fused, verified, and applied through the knowledge service. Progress is measured +before another bounded iteration is considered. + +```bash +nexus-research create "What is Acme's verified status?" \ + --criterion "two independent sources support the answer" +nexus-research status SESSION_ID +nexus-research explain SESSION_ID +nexus-investigation-bench +``` + +See [autonomous investigation](docs/autonomous-investigation.md) and the +[closed epistemic loop](docs/autonomous-loop.md). + **Quick start** ```bash diff --git a/docs/adr/0003-evidence-verification-boundary.md b/docs/adr/0004-evidence-verification-boundary.md similarity index 95% rename from docs/adr/0003-evidence-verification-boundary.md rename to docs/adr/0004-evidence-verification-boundary.md index 8d63849..542be3e 100644 --- a/docs/adr/0003-evidence-verification-boundary.md +++ b/docs/adr/0004-evidence-verification-boundary.md @@ -1,4 +1,4 @@ -# ADR 0003: verify evidence before promoting knowledge +# ADR 0004: verify evidence before promoting knowledge ## Context diff --git a/docs/adr/0005-ready-wave-runtime-integration.md b/docs/adr/0005-ready-wave-runtime-integration.md new file mode 100644 index 0000000..4593104 --- /dev/null +++ b/docs/adr/0005-ready-wave-runtime-integration.md @@ -0,0 +1,26 @@ +# ADR 0005: Submit investigation DAGs in dependency-ready waves + +## Status + +Accepted for Phase 4. + +## Context + +`InvestigationPlan` needs dependencies, while the Phase 3 `DistributedTask` contract +intentionally owns placement and delivery but has no dependency field. Adding a second +scheduler in the investigation layer or reaching into coordinator storage would break +the subsystem boundary. + +## Decision + +Plans compile through the existing `TaskDAG`. `PlanExecutionController` persists the +plan-to-AgentRun-to-distributed-task mapping and submits only nodes whose parent tasks +have succeeded. The normal `RuntimeApplication.submit_task` boundary remains the only +submission path. Capabilities, priorities, retries, worker assignment, leases, +cancellation, and recovery remain owned by the distributed runtime. + +## Consequences + +Independent nodes are queued together and remain parallelizable. Dependent nodes are +submitted in later waves. Permanent parent failure blocks descendants explicitly. +Session restart can reconstruct the mapping without depending on Python object identity. diff --git a/docs/api-contract.md b/docs/api-contract.md index badeb12..d8d99a3 100644 --- a/docs/api-contract.md +++ b/docs/api-contract.md @@ -11,3 +11,11 @@ adapters call these operations without exposing domain internals: The initial `RuntimeAPI` exposes them in-process. Adapters propagate trace/correlation IDs, authenticate callers, authorize cancellation and creation, and translate `DomainError` to their transport's validation response. + +Phase 4 adds a separate `InvestigationApplication` boundary. Its explicit stage +operations are `create`, `observe`, `generate`, `score`, `select`, `build_plan`, +`start_execution`, `advance_execution`, `collect_execution_results`, +`collect_evidence`, `evaluate`, `verify`, `update_knowledge`, and +`finish_iteration`. `status`, `pause`, `resume`, `cancel`, `explain`, and +`resume_iteration` provide lifecycle and recovery operations. The CLI delegates to +this boundary and contains no planning, execution, or verification policy. diff --git a/docs/autonomous-investigation.md b/docs/autonomous-investigation.md new file mode 100644 index 0000000..fcf795b --- /dev/null +++ b/docs/autonomous-investigation.md @@ -0,0 +1,105 @@ +# Autonomous Investigation Engine + +Phase 4 adds the intelligence layer between the existing knowledge engine and the +existing distributed AgentRun runtime. It does not replace retrieval, GraphRAG, gap +analysis, the Agent Harness, task leases, retries, or worker scheduling. + +```mermaid +flowchart LR + K1["Knowledge Intelligence"] -->|"snapshot + KnowledgeGap"| I["Investigation Engine"] + I -->|"InvestigationPlan + ready task wave"| D["Distributed Runtime"] + D -->|"AgentRun"| H["Agent Harness"] + H -->|"InvestigationResult + Evidence"| I + I -->|"verified KnowledgeUpdate"| K1 +``` + +## Public domain model + +- `ResearchObjective` records the question, scope, constraints, success criteria, + metadata, identifier, and creation time. +- `InvestigationSession` is an explicit `PLANNING → EXECUTING → EVALUATING → + UPDATING` state machine with paused and terminal states. +- `KnowledgeSnapshot` captures retrieval references, GraphRAG entities and relations, + gaps, contradiction keys, uncertainty, and observation metadata. +- `CandidateInvestigation` is a deterministic transformation of a measurable gap. It + records the hypothesis, evidence need, capability requirements, gain, cost, time, + risk, and priority. +- `InvestigationPlan` validates dependencies and compiles through the existing + `TaskDAG` and distributed task contracts. +- `Evidence`, `Evaluation`, `VerificationReport`, `InvestigationKnowledgeUpdate`, and + `ProgressReport` make the epistemic result inspectable rather than concatenating + agent text. + +All identifiers and important state are serializable. `SQLiteInvestigationRepository` +persists the objective, session, iteration, and ordered artifacts. A process restart +can reload collected `InvestigationResult` and `EvidenceSet` values and continue +evaluation without rerunning completed agents. + +## Application boundary + +`InvestigationApplication` exposes independently callable stages: + +1. `observe` +2. `generate` +3. `score` +4. `select` +5. `build_plan` +6. `start_execution` / `advance_execution` +7. `collect_evidence` +8. `evaluate` +9. `verify` +10. `update_knowledge` +11. `finish_iteration` + +`plan_iteration` and `process_results` are thin compositions of those services. They +do not hide an unbounded research loop. + +The high-level CLI is also thin: + +```text +nexus-research create QUESTION --criterion CRITERION +nexus-research status SESSION_ID +nexus-research pause SESSION_ID +nexus-research resume SESSION_ID +nexus-research cancel SESSION_ID +nexus-research explain SESSION_ID +nexus-research evidence SESSION_ID +nexus-research gaps SESSION_ID +nexus-research plan SESSION_ID +nexus-research iterations SESSION_ID +``` + +## Execution contract + +The current distributed record intentionally has no dependency field. The +investigation layer therefore submits only dependency-ready waves. It never assigns a +worker, claims a lease, retries an attempt, or executes an agent. When a parent task +succeeds, `PlanExecutionController` submits newly ready children through +`RuntimeApplication`. A cancelled or dead-lettered parent marks descendants blocked +in the persisted plan execution rather than pretending they ran. + +The first wave can contain many independent tasks, allowing the existing scheduler to +place them concurrently according to required capabilities. + +## Explainability and observability + +`explain` reconstructs three decision chains: + +- gap → gain/importance/uncertainty/cost/risk → score → selection; +- claim → evidence → source lineage → evaluation → verification; +- termination reason → objective confidence → gaps → contradictions → budget. + +Structured events use the `investigation.*` namespace and preserve objective/session +correlation. Metrics include sessions, iterations, gaps discovered and resolved, +generated/executed investigations, evidence, contradictions, updates, information +gain, and cost per resolved gap. + +Run the deterministic 10/50/10 baseline with: + +```bash +nexus-investigation-bench +``` + +See [lifecycle](investigation-lifecycle.md), [planning](investigation-planning.md), +[evidence](evidence-model.md), [provenance](provenance.md), +[verification](verification.md), and [closed loop](autonomous-loop.md). diff --git a/docs/autonomous-loop.md b/docs/autonomous-loop.md new file mode 100644 index 0000000..e3cda0a --- /dev/null +++ b/docs/autonomous-loop.md @@ -0,0 +1,91 @@ +# Closed Epistemic Loop + +The investigation engine advances one explicit stage at a time. There is no +`while True` research loop and no prompt that substitutes for control flow. + +```mermaid +stateDiagram-v2 + [*] --> PLANNING + PLANNING --> EXECUTING: selected plan + EXECUTING --> EVALUATING: complete or accepted partial results + EVALUATING --> UPDATING: verification report + UPDATING --> PLANNING: continue after progress measurement + PLANNING --> COMPLETED: no valuable work / objective satisfied + UPDATING --> COMPLETED: stopping criterion + PLANNING --> PAUSED + EXECUTING --> PAUSED + EVALUATING --> PAUSED + UPDATING --> PAUSED + PAUSED --> PLANNING + PAUSED --> EXECUTING + PAUSED --> EVALUATING + PAUSED --> UPDATING + PLANNING --> CANCELLED + EXECUTING --> CANCELLED + EVALUATING --> FAILED + UPDATING --> FAILED +``` + +## One iteration + +```mermaid +flowchart TD + O["Observe retrieval, GraphRAG, gaps, contradictions"] --> G["Generate grounded candidates"] + G --> S["Score gain, importance, uncertainty, availability, cost, time, risk, redundancy"] + S --> P["Budget/capacity selection and TaskDAG plan"] + P --> R["Submit ready AgentRuns to distributed runtime"] + R --> E["Collect complete and partial evidence with lineage"] + E --> F["Fuse duplicates, support, contradictions, and low-quality evidence"] + F --> V["Verify provenance, independent sources, quality, confidence, conflict policy"] + V --> U["Commit eligible updates through KnowledgeEngine"] + U --> N["Observe new knowledge and calculate progress"] + N --> T{"Termination policy"} + T -->|"continue"| O + T -->|"bounded stop"| X["Completed / Cancelled / Failed"] +``` + +## Bounded termination + +Every session has maximum iterations, investigations, AgentRuns, cost, and execution +time. The deterministic termination policy handles objective satisfaction, confidence, +budget exhaustion, maximum iterations, no valuable candidate, unresolvable conflict, +user cancellation, and system failure in a documented precedence order. + +An uncertain result is valid. Verification preserves `confirmed`, `probable`, +`uncertain`, `contradicted`, and `insufficient_evidence`; only policy-eligible claims +reach the knowledge-update boundary. + +## Partial results and failure + +Distributed retry and lease recovery stay in the coordinator. The investigation layer +sees the eventual task state and can evaluate completed or partial +`InvestigationResult` values while other tasks remain running. Failed dependencies are +recorded as blocked investigations. They are not silently counted as success. + +Collected results are persisted before evaluation. After restart, +`resume_collected_iteration` reloads the evidence, continues evaluation and update, and +does not invoke the Agent Harness again. At-least-once distributed execution can still +duplicate delivery; evidence fingerprints and stable result lineage make fusion +idempotent and inspectable. + +`resume_iteration` also reconstructs sessions interrupted during `EVALUATING` or +`UPDATING`. The update proposal is persisted before calling the knowledge service. If +the process fails after that call but before the receipt is stored, recovery may submit +the same proposal again. Claim and evidence identifiers remain stable, so repository +adapters must provide idempotent save-by-ID behavior. This is an explicit at-least-once +knowledge-update boundary; the application does not claim a distributed transaction +across its SQLite record and an external knowledge store. + +## Progress + +`ProgressMeasurer` compares before/after gap and contradiction snapshots and reports: + +- gaps resolved, remaining, and newly discovered; +- uncertainty reduced; +- contradictions introduced and resolved; +- evidence and updates; +- information gain; +- cost per resolved gap. + +This report is the input to the next stopping decision and is persisted with the full +iteration history. diff --git a/docs/distributed-execution.md b/docs/distributed-execution.md index e6ae8f6..e5b7025 100644 --- a/docs/distributed-execution.md +++ b/docs/distributed-execution.md @@ -10,3 +10,9 @@ must exist and the graph is validated for cycles. Independent ready tasks can le different workers concurrently. The scheduler enforces backpressure, worker concurrency, timeouts, retry policy, cancellation, and failed-dependency cancellation. On restart it reloads snapshots and deliberately reclaims former worker leases. + +The worker invokes the Agent Harness with an immutable `HarnessExecutionContext`: +`run_id`, `correlation_id`, `task_id`, `attempt_id`, `lease_id`, `worker_id`, and the +task metadata produced by the investigation plan. Harness implementations use this +public context to construct evidence lineage; they do not query coordinator storage or +infer the active attempt from in-process worker state. diff --git a/docs/events.md b/docs/events.md index bac6eb6..b1795f2 100644 --- a/docs/events.md +++ b/docs/events.md @@ -5,6 +5,27 @@ trace ID, correlation ID, optional causation ID, and structured payload. Stable namespaces are `agent`, `task`, `tool`, `worker`, `investigation`, `hypothesis`, `experiment`, and `knowledge_update`. +The autonomous investigation application emits these stable lifecycle events: + +- `investigation.session_created` +- `investigation.planning_started` +- `investigation.gaps_identified` +- `investigation.candidates_generated` +- `investigation.selected` +- `investigation.plan_created` +- `investigation.execution_started` +- `investigation.evidence_collected` +- `investigation.evaluated` +- `investigation.verification_started` +- `investigation.knowledge_updated` +- `investigation.iteration_completed` +- `investigation.completed` +- `investigation.failed` + +Their payloads include the session, objective, and iteration. The session is the trace +identifier and the objective is the correlation identifier, preserving a stable chain +from planning through distributed execution and knowledge update. + `EventBus` is broker-independent: `publish`, `subscribe`, `acknowledge`, and `dead_letter`. `InMemoryEventBus` is deterministic for tests. Production adapters must define consumer-group, retention, authentication, ordering, acknowledgement, retry, and diff --git a/docs/leases.md b/docs/leases.md index b9d0cab..d4dd455 100644 --- a/docs/leases.md +++ b/docs/leases.md @@ -17,7 +17,7 @@ sequenceDiagram TaskStore-->>Worker: attempt_id + lease_id Worker->>Coordinator: start(lease_id) Worker->>Coordinator: heartbeat + renew - Worker->>Harness: execute_or_resume(run_id) + Worker->>Harness: execute_or_resume(HarnessExecutionContext) Worker->>Coordinator: complete(lease_id) Coordinator->>TaskStore: atomic completion ``` diff --git a/docs/provenance.md b/docs/provenance.md index f937154..0db27bc 100644 --- a/docs/provenance.md +++ b/docs/provenance.md @@ -30,3 +30,8 @@ knowledge subsystem's chunk, document, source, and evidence identifiers. The knowledge update adapter never writes repositories or graph storage. It builds the existing `KnowledgeUpdate` service contract and submits it with `KnowledgeEngine.commit_knowledge_update`. + +Before submission, `KnowledgeEngine.validate_evidence_provenance` verifies that the +referenced source, document, and chunk exist, form the declared source → document → +chunk chain, and that the source reference matches. A structurally complete but +dangling lineage is rejected rather than being written as trusted knowledge. diff --git a/pyproject.toml b/pyproject.toml index c19834d..76c3ef0 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "nexus" -version = "0.3.0" +version = "0.4.0" description = "NEXUS autonomous knowledge-discovery platform: knowledge intelligence engine and fault-tolerant research runtime" readme = "README.md" requires-python = ">=3.11" @@ -31,6 +31,8 @@ dev = ["ruff>=0.6", "mypy>=1.11", "pytest>=7.4", "pytest-cov>=4.1"] nexus-knowledge = "nexus_knowledge.cli:main" nexus-runtime = "nexus_runtime.distributed.cli:main" nexus-runtime-bench = "nexus_runtime.distributed.benchmark:main" +nexus-research = "nexus_runtime.investigation.cli:main" +nexus-investigation-bench = "nexus_runtime.investigation.benchmark:main" [tool.setuptools.packages.find] where = ["src"] @@ -59,3 +61,7 @@ select = ["E", "F", "I", "UP", "B"] python_version = "3.11" strict = true files = ["src/nexus_runtime"] + +[[tool.mypy.overrides]] +module = ["nexus_knowledge.*", "numpy.*"] +follow_imports = "skip" diff --git a/src/nexus_knowledge/service/engine.py b/src/nexus_knowledge/service/engine.py index 9c758f6..e296b76 100644 --- a/src/nexus_knowledge/service/engine.py +++ b/src/nexus_knowledge/service/engine.py @@ -271,6 +271,26 @@ def provenance(self, claim_id: str) -> ProvenanceResponse: source_references={sid: sources[sid].reference for sid in source_ids if sid in sources}, ) + def validate_evidence_provenance( + self, + source_id: str, + document_id: str, + chunk_id: str, + source_reference: str, + ) -> bool: + """Validate a source → document → chunk chain through the public service boundary.""" + source = self.repository.sources.get(source_id) + document = self.repository.documents.get(document_id) + chunk = self.repository.chunks.get(chunk_id) + return bool( + source is not None + and document is not None + and chunk is not None + and source.reference == source_reference + and document.source_id == source.id + and chunk.document_id == document.id + ) + # -- transactions --------------------------------------------------- def commit_knowledge_update(self, update: KnowledgeUpdate) -> KnowledgeUpdateReceipt: """Atomically apply a batch of knowledge changes.""" diff --git a/src/nexus_runtime/distributed/__init__.py b/src/nexus_runtime/distributed/__init__.py index 685beaf..e527cb6 100644 --- a/src/nexus_runtime/distributed/__init__.py +++ b/src/nexus_runtime/distributed/__init__.py @@ -15,7 +15,7 @@ TaskPriority, ) from .store import InMemoryTaskStore, LeaseStore, SQLiteTaskStore, TaskStore -from .worker import Harness, HarnessOutcome, Worker +from .worker import Harness, HarnessExecutionContext, HarnessOutcome, Worker __all__ = [ "Clock", @@ -24,6 +24,7 @@ "DistributedTaskState", "FailureClass", "Harness", + "HarnessExecutionContext", "HarnessOutcome", "InMemoryTaskStore", "Lease", diff --git a/src/nexus_runtime/distributed/simulator.py b/src/nexus_runtime/distributed/simulator.py index 386ef84..fd28382 100644 --- a/src/nexus_runtime/distributed/simulator.py +++ b/src/nexus_runtime/distributed/simulator.py @@ -14,7 +14,7 @@ from .model import TERMINAL_TASK_STATES, DistributedTask, TaskPriority from .security import WorkerIdentity from .store import InMemoryTaskStore -from .worker import Harness, HarnessOutcome, HarnessStatus, Worker +from .worker import Harness, HarnessExecutionContext, HarnessOutcome, HarnessStatus, Worker class DeterministicHarness: @@ -28,16 +28,15 @@ def __init__( def execute_or_resume( self, - run_id: str, - correlation_id: str, + context: HarnessExecutionContext, cancellation_requested: Callable[[], bool], ) -> HarnessOutcome: - self.calls.append(run_id) + self.calls.append(context.run_id) if self._execute: - return self._execute(run_id, correlation_id, cancellation_requested) + return self._execute(context.run_id, context.correlation_id, cancellation_requested) if cancellation_requested(): return HarnessOutcome(HarnessStatus.CANCELLED) - return HarnessOutcome(HarnessStatus.SUCCEEDED, result_ref=f"result://{run_id}") + return HarnessOutcome(HarnessStatus.SUCCEEDED, result_ref=f"result://{context.run_id}") def cancel_run(self, run_id: str) -> None: self.cancelled.append(run_id) diff --git a/src/nexus_runtime/distributed/worker.py b/src/nexus_runtime/distributed/worker.py index 0883e73..f1bd9bc 100644 --- a/src/nexus_runtime/distributed/worker.py +++ b/src/nexus_runtime/distributed/worker.py @@ -2,12 +2,13 @@ from __future__ import annotations -from collections.abc import Callable +from collections.abc import Callable, Mapping from dataclasses import dataclass from enum import StrEnum from threading import Event as ThreadEvent from threading import Thread -from typing import Protocol +from types import MappingProxyType +from typing import Any, Protocol from ..models import DomainError from .coordinator import Coordinator @@ -30,13 +31,25 @@ class HarnessOutcome: error: str = "" +@dataclass(frozen=True, slots=True) +class HarnessExecutionContext: + """Immutable lineage supplied by the worker to the Agent Harness.""" + + run_id: str + correlation_id: str + task_id: str + attempt_id: str + lease_id: str + worker_id: str + metadata: Mapping[str, Any] + + class Harness(Protocol): """Small integration port; implementations own all AgentRun execution details.""" def execute_or_resume( self, - run_id: str, - correlation_id: str, + context: HarnessExecutionContext, cancellation_requested: Callable[[], bool], ) -> HarnessOutcome: ... @@ -101,7 +114,16 @@ def cancellation_requested() -> bool: try: outcome = self._harness.execute_or_resume( - task.run_id, task.correlation_id, cancellation_requested + HarnessExecutionContext( + run_id=task.run_id, + correlation_id=task.correlation_id, + task_id=task.task_id, + attempt_id=attempt_id, + lease_id=lease_id, + worker_id=self.identity.worker_id, + metadata=MappingProxyType(dict(task.metadata)), + ), + cancellation_requested, ) except Exception as exc: heartbeat_stop.set() diff --git a/src/nexus_runtime/investigation/__init__.py b/src/nexus_runtime/investigation/__init__.py index 0e1f808..ef1b46c 100644 --- a/src/nexus_runtime/investigation/__init__.py +++ b/src/nexus_runtime/investigation/__init__.py @@ -1 +1,47 @@ -"""Autonomous-investigation domain and planning services.""" +"""Autonomous-investigation domain, planning, evidence, and application services.""" + +from .application import InvestigationApplication, IterationOutcome, PlanningOutcome +from .evidence import Evidence, EvidenceSet, InvestigationResult +from .execution import PlanExecution, PlanExecutionController +from .generator import CandidateInvestigation, InvestigationGenerator +from .objective import ResearchObjective +from .planner import InvestigationPlan, InvestigationPlanner +from .repository import ( + InMemoryInvestigationRepository, + InvestigationRecord, + SQLiteInvestigationRepository, +) +from .results import ( + InMemoryInvestigationResultRepository, + RuntimeResultCollector, + SQLiteInvestigationResultRepository, +) +from .session import InvestigationBudget, InvestigationSession, SessionState +from .verification import EpistemicStatus, VerificationPolicy + +__all__ = [ + "CandidateInvestigation", + "EpistemicStatus", + "Evidence", + "EvidenceSet", + "InMemoryInvestigationRepository", + "InvestigationApplication", + "InvestigationBudget", + "InvestigationGenerator", + "InvestigationPlan", + "InvestigationPlanner", + "InvestigationRecord", + "InvestigationResult", + "InvestigationSession", + "IterationOutcome", + "InMemoryInvestigationResultRepository", + "PlanExecution", + "PlanExecutionController", + "PlanningOutcome", + "ResearchObjective", + "RuntimeResultCollector", + "SQLiteInvestigationRepository", + "SQLiteInvestigationResultRepository", + "SessionState", + "VerificationPolicy", +] diff --git a/src/nexus_runtime/investigation/application.py b/src/nexus_runtime/investigation/application.py new file mode 100644 index 0000000..0f7e117 --- /dev/null +++ b/src/nexus_runtime/investigation/application.py @@ -0,0 +1,990 @@ +"""Explicit application services for the closed-loop investigation lifecycle.""" + +from __future__ import annotations + +import json +from collections.abc import Mapping, Sequence +from dataclasses import asdict, dataclass, is_dataclass +from datetime import datetime, timedelta +from enum import Enum +from typing import Any, Protocol, cast + +from nexus_runtime.events import Event, EventBus, InMemoryEventBus +from nexus_runtime.models import DomainError, utcnow + +from .evaluation import Evaluation, EvidenceEvaluator +from .evidence import EvidenceSet, InvestigationResult, InvestigationResultState +from .execution import ExecutionStatus, PlanExecution, PlanExecutionController +from .generator import CandidateInvestigation, InvestigationGenerator, KnowledgeGapLike +from .knowledge_update import ( + InvestigationKnowledgeUpdate, + KnowledgeUpdateIntegrator, + KnowledgeUpdatePort, + KnowledgeUpdateResult, +) +from .metrics import InMemoryInvestigationMetrics, InvestigationMetrics +from .objective import ResearchObjective +from .observation import KnowledgeSnapshot +from .planner import InvestigationPlan, InvestigationPlanner +from .progress import GapState, ProgressMeasurer, ProgressReport +from .repository import ( + InMemoryInvestigationRepository, + InvestigationRecord, + InvestigationRepository, +) +from .results import RuntimeResultCollector +from .scoring import InvestigationScore, InvestigationScoringModel +from .selector import InvestigationSelector, SelectionResult +from .session import InvestigationBudget, InvestigationSession, SessionState, TerminationReason +from .termination import TerminationContext, TerminationDecision, TerminationPolicy +from .verification import ClaimVerifier, VerificationReport + + +class KnowledgeObservationPort(Protocol): + """Read-only public knowledge operations used to construct a snapshot.""" + + def retrieve(self, query: str, top_k: int = 10) -> Any: ... + + def graphrag(self, query: str, top_k: int = 8, depth: int = 2) -> Any: ... + + def find_knowledge_gaps(self) -> list[Any]: ... + + def detect_contradictions(self) -> list[Any]: ... + + +class KnowledgeObserver: + """Observe retrieval, GraphRAG, uncertainty gaps, and contradictions via one boundary.""" + + def __init__(self, knowledge: KnowledgeObservationPort) -> None: + self._knowledge = knowledge + + def observe( + self, objective: ResearchObjective, *, at: datetime | None = None + ) -> KnowledgeSnapshot: + retrieval = self._knowledge.retrieve(objective.question, top_k=10) + evidence_graph = self._knowledge.graphrag(objective.question, top_k=8, depth=2) + gaps = cast("Sequence[KnowledgeGapLike]", self._knowledge.find_knowledge_gaps()) + contradictions = self._knowledge.detect_contradictions() + retrieval_refs = tuple( + str(getattr(item, "chunk_id", "")) + for item in getattr(retrieval, "candidates", ()) + if getattr(item, "chunk_id", "") + ) + entity_ids = tuple( + str(getattr(item, "id", "")) + for item in getattr(evidence_graph, "entities", ()) + if getattr(item, "id", "") + ) + relation_ids = tuple( + str(getattr(item, "id", "")) + for item in getattr(evidence_graph, "relations", ()) + if getattr(item, "id", "") + ) + contradiction_keys = tuple(self._contradiction_key(item) for item in contradictions) + confidence = float(getattr(evidence_graph, "confidence", 0.0)) + return KnowledgeSnapshot.capture( + objective, + gaps, + observed_at=at or utcnow(), + retrieval_refs=retrieval_refs, + entity_ids=entity_ids, + relation_ids=relation_ids, + contradiction_ids=contradiction_keys, + summary=( + f"{len(gaps)} gaps, {len(contradictions)} contradictions, " + f"GraphRAG confidence {confidence:.3f}" + ), + metadata={ + "retrieval_request_id": str(getattr(retrieval, "request_id", "")), + "graphrag_confidence": confidence, + }, + ) + + @staticmethod + def _contradiction_key(contradiction: Any) -> str: + kind = str(getattr(contradiction, "kind", "contradiction")) + pair = sorted( + ( + str(getattr(contradiction, "claim_a_id", "unknown-a")), + str(getattr(contradiction, "claim_b_id", "unknown-b")), + ) + ) + return f"{kind}:{pair[0]}:{pair[1]}" + + +@dataclass(frozen=True, slots=True) +class PlanningOutcome: + snapshot: KnowledgeSnapshot + candidates: tuple[CandidateInvestigation, ...] + scores: tuple[InvestigationScore, ...] + selection: SelectionResult + plan: InvestigationPlan | None + termination: TerminationDecision | None = None + + +@dataclass(frozen=True, slots=True) +class IterationOutcome: + evidence_set: EvidenceSet + evaluation: Evaluation + verification: VerificationReport + knowledge_update: InvestigationKnowledgeUpdate + update_result: KnowledgeUpdateResult + progress: ProgressReport + termination: TerminationDecision + + +class InvestigationApplication: + """Transport-independent API; each epistemic stage remains separately callable.""" + + def __init__( + self, + knowledge: KnowledgeObservationPort, + *, + knowledge_updates: KnowledgeUpdatePort | None = None, + repository: InvestigationRepository | None = None, + execution: PlanExecutionController | None = None, + result_collector: RuntimeResultCollector | None = None, + event_bus: EventBus | None = None, + metrics: InvestigationMetrics | None = None, + generator: InvestigationGenerator | None = None, + scoring: InvestigationScoringModel | None = None, + selector: InvestigationSelector | None = None, + planner: InvestigationPlanner | None = None, + evaluator: EvidenceEvaluator | None = None, + verifier: ClaimVerifier | None = None, + termination: TerminationPolicy | None = None, + ) -> None: + self._observer = KnowledgeObserver(knowledge) + self._repository = repository or InMemoryInvestigationRepository() + self._execution = execution + self._result_collector = result_collector + self._events = event_bus or InMemoryEventBus() + self._metrics = metrics or InMemoryInvestigationMetrics() + self._generator = generator or InvestigationGenerator() + self._scoring = scoring or InvestigationScoringModel() + self._selector = selector or InvestigationSelector(minimum_score=0.05) + self._planner = planner or InvestigationPlanner() + self._evaluator = evaluator or EvidenceEvaluator() + self._verifier = verifier or ClaimVerifier() + self._termination = termination or TerminationPolicy() + update_boundary = knowledge_updates + if update_boundary is None: + update_boundary = cast(KnowledgeUpdatePort, knowledge) + self._updates = KnowledgeUpdateIntegrator(update_boundary) + + def create( + self, objective: ResearchObjective, budget: InvestigationBudget + ) -> InvestigationSession: + session = InvestigationSession(objective_id=objective.objective_id, budget=budget) + record = InvestigationRecord(objective, session) + self._repository.save(record) + self._metrics.increment("investigation_sessions") + self._emit("investigation.session_created", record, {"question": objective.question}) + return session + + def status(self, session_id: str) -> InvestigationRecord: + return self._record(session_id) + + def observe(self, session_id: str) -> KnowledgeSnapshot: + record = self._record(session_id) + if record.session.state not in {SessionState.PLANNING, SessionState.UPDATING}: + raise DomainError("knowledge can be observed only while PLANNING or UPDATING") + self._emit("investigation.planning_started", record, {}) + snapshot = self._observer.observe(record.objective) + record.append("knowledge_snapshot", snapshot.to_dict()) + self._repository.save(record) + self._metrics.increment("gaps_discovered", len(snapshot.gaps)) + self._emit( + "investigation.gaps_identified", + record, + {"snapshot_id": snapshot.snapshot_id, "count": len(snapshot.gaps)}, + ) + return snapshot + + def generate( + self, session_id: str, snapshot: KnowledgeSnapshot + ) -> tuple[CandidateInvestigation, ...]: + record = self._record(session_id) + self._require_snapshot(record, snapshot) + candidates = self._generator.generate(record.objective, snapshot.gaps) + record.append( + "candidate_investigations", {"items": [item.to_dict() for item in candidates]} + ) + self._repository.save(record) + self._metrics.increment("investigations_generated", len(candidates)) + self._emit("investigation.candidates_generated", record, {"count": len(candidates)}) + return candidates + + def score( + self, + session_id: str, + snapshot: KnowledgeSnapshot, + candidates: Sequence[CandidateInvestigation], + ) -> tuple[InvestigationScore, ...]: + record = self._record(session_id) + self._require_snapshot(record, snapshot) + candidate_artifact = self._required_artifact(record, "candidate_investigations") + if ( + candidate_artifact.iteration != record.session.iteration + or candidate_artifact.payload.get("items") != [item.to_dict() for item in candidates] + ): + raise DomainError("candidate investigations are not active for this iteration") + scores = self._scoring.score_all(candidates, snapshot.gaps) + forecast = self._scoring.forecast(scores) + record.append( + "investigation_scores", + { + "items": [item.to_dict() for item in scores], + "forecast": forecast.to_dict(), + }, + ) + self._repository.save(record) + return scores + + def select( + self, + session_id: str, + scores: Sequence[InvestigationScore], + *, + worker_capacity: int, + top_k: int | None = None, + ) -> SelectionResult: + record = self._record(session_id) + score_artifact = self._required_artifact(record, "investigation_scores") + if score_artifact.iteration != record.session.iteration or score_artifact.payload.get( + "items" + ) != [item.to_dict() for item in scores]: + raise DomainError("investigation scores are not active for this iteration") + selection = self._selector.select( + scores, + budget=record.session.budget, + usage=record.session.usage, + worker_capacity=worker_capacity, + top_k=top_k, + ) + record.append("investigation_selection", selection.to_dict()) + self._repository.save(record) + self._emit( + "investigation.selected", + record, + { + "selected": [item.candidate.investigation_id for item in selection.selected], + "rejected": selection.rejected, + }, + ) + return selection + + def build_plan( + self, + session_id: str, + selection: SelectionResult, + *, + dependencies: Mapping[str, tuple[str, ...]] | None = None, + ) -> InvestigationPlan: + record = self._record(session_id) + selection_artifact = self._required_artifact(record, "investigation_selection") + if ( + selection_artifact.iteration != record.session.iteration + or selection_artifact.payload != selection.to_dict() + ): + raise DomainError("investigation selection is not active for this iteration") + plan = self._planner.build(record.session, selection, dependencies=dependencies) + record.append("investigation_plan", plan.to_dict()) + self._repository.save(record) + self._emit("investigation.plan_created", record, {"plan_id": plan.plan_id}) + return plan + + def plan_iteration( + self, + session_id: str, + *, + worker_capacity: int, + top_k: int | None = None, + dependencies: Mapping[str, tuple[str, ...]] | None = None, + ) -> PlanningOutcome: + """Thin composition of the independently callable planning stages.""" + snapshot = self.observe(session_id) + candidates = self.generate(session_id, snapshot) + scores = self.score(session_id, snapshot, candidates) + selection = self.select(session_id, scores, worker_capacity=worker_capacity, top_k=top_k) + if selection.selected: + plan = self.build_plan(session_id, selection, dependencies=dependencies) + return PlanningOutcome(snapshot, candidates, scores, selection, plan) + record = self._record(session_id) + decision = self._termination.evaluate( + record.session, + TerminationContext( + objective_satisfied=not snapshot.gaps, + objective_confidence=float(snapshot.metadata.get("graphrag_confidence", 0.0)), + remaining_gap_count=len(snapshot.gaps), + best_candidate_score=None, + ), + ) + self._apply_termination(record, decision) + return PlanningOutcome(snapshot, candidates, scores, selection, None, decision) + + def start_execution( + self, + session_id: str, + plan: InvestigationPlan, + run_ids: Mapping[str, str], + ) -> ExecutionStatus: + record = self._record(session_id) + if self._execution is None: + raise DomainError("distributed execution adapter is not configured") + if record.session.state != SessionState.PLANNING or plan.session_id != session_id: + raise DomainError("session is not ready to execute this plan") + plan_artifact = self._required_artifact(record, "investigation_plan") + if ( + plan_artifact.iteration != record.session.iteration + or plan_artifact.payload != plan.to_dict() + ): + raise DomainError("only the current iteration plan can be executed") + if any( + artifact.kind == "plan_execution" + and artifact.iteration == record.session.iteration + and artifact.payload.get("plan_id") == plan.plan_id + for artifact in record.artifacts + ): + raise DomainError("investigation plan has already been started") + investigation_cost = sum(item.estimated_cost for item in plan.investigations) + remaining = record.session.remaining_budget() + if ( + len(plan.investigations) > remaining["investigations"] + or len(run_ids) > remaining["agent_runs"] + or investigation_cost > remaining["cost"] + ): + raise DomainError("investigation plan exceeds remaining session budget") + execution = self._execution.prepare(plan, run_ids) + record.session.transition(SessionState.EXECUTING) + record.session.record_usage( + investigations=len(plan.investigations), + agent_runs=len(run_ids), + cost=investigation_cost, + ) + record.append("plan_execution", execution.to_dict()) + self._repository.save(record) + status = self._execution.advance(plan, execution) + record.append("plan_execution", status.execution.to_dict()) + self._repository.save(record) + self._metrics.increment("investigations_executed", len(plan.investigations)) + self._emit( + "investigation.execution_started", + record, + {"plan_id": plan.plan_id, "task_ids": status.execution.task_ids}, + ) + return status + + def advance_execution(self, session_id: str) -> ExecutionStatus: + record = self._record(session_id) + if self._execution is None: + raise DomainError("distributed execution adapter is not configured") + if record.session.state != SessionState.EXECUTING: + raise DomainError("distributed execution can advance only while EXECUTING") + plan_artifact = self._required_artifact(record, "investigation_plan") + execution_artifact = self._required_artifact(record, "plan_execution") + plan = InvestigationPlan.from_dict(plan_artifact.payload) + execution = PlanExecution.from_dict(execution_artifact.payload) + status = self._execution.advance(plan, execution) + record.append("plan_execution", status.execution.to_dict()) + self._repository.save(record) + return status + + def collect_execution_results(self, session_id: str) -> tuple[InvestigationResult, ...]: + if self._result_collector is None: + raise DomainError("distributed result collector is not configured") + record = self._record(session_id) + execution = PlanExecution.from_dict( + self._required_artifact(record, "plan_execution").payload + ) + return self._result_collector.collect(execution) + + def collect_evidence( + self, + session_id: str, + results: Sequence[InvestigationResult], + ) -> EvidenceSet: + record = self._record(session_id) + if record.session.state != SessionState.EXECUTING: + raise DomainError("evidence can be collected only while EXECUTING") + if any(item.session_id != session_id for item in results): + raise DomainError("investigation result belongs to another session") + if any( + item.kind == "evidence_set" and item.iteration == record.session.iteration + for item in record.artifacts + ): + raise DomainError("evidence has already been collected for this iteration") + execution = PlanExecution.from_dict( + self._required_artifact(record, "plan_execution").payload + ) + self._validate_results(execution, results) + if self._result_collector is None: + raise DomainError("distributed result collector is required for lineage validation") + authoritative = {item.result_id: item for item in self._result_collector.collect(execution)} + if any( + item.result_id not in authoritative + or item.to_dict() != authoritative[item.result_id].to_dict() + for item in results + ): + raise DomainError("investigation result is not the durable runtime outcome") + completed = tuple( + item + for item in results + if item.state in {InvestigationResultState.COMPLETED, InvestigationResultState.PARTIAL} + ) + successful = sum(item.state == InvestigationResultState.COMPLETED for item in results) + failed = sum( + item.state in {InvestigationResultState.FAILED, InvestigationResultState.CANCELLED} + for item in results + ) + evidence = tuple(item for result in completed for item in result.evidence_set.evidence) + evidence_set = EvidenceSet(session_id=session_id, evidence=evidence) + record.append("investigation_results", {"items": [item.to_dict() for item in results]}) + record.append("evidence_set", evidence_set.to_dict()) + if results: + completed_at = max(item.completed_at for item in results) + elapsed = max(timedelta(0), completed_at - record.session.updated_at) + record.session.record_usage(execution_time=elapsed) + self._repository.save(record) + self._metrics.increment("evidence_collected", len(evidence)) + self._metrics.increment("investigation_successes", successful) + self._metrics.increment("investigation_failures", failed) + self._emit( + "investigation.evidence_collected", + record, + {"evidence_set_id": evidence_set.evidence_set_id, "count": len(evidence)}, + ) + return evidence_set + + def stored_results(self, session_id: str) -> tuple[InvestigationResult, ...]: + """Rehydrate durable worker results after an application restart.""" + record = self._record(session_id) + artifact = self._required_artifact(record, "investigation_results") + items = artifact.payload.get("items") + if not isinstance(items, list) or any(not isinstance(item, dict) for item in items): + raise DomainError("persisted investigation results are malformed") + try: + return tuple(InvestigationResult.from_dict(item) for item in items) + except ValueError as exc: + raise DomainError("persisted investigation results are malformed") from exc + + def evaluate(self, session_id: str, evidence_set: EvidenceSet) -> Evaluation: + record = self._record(session_id) + if evidence_set.session_id != session_id: + raise DomainError("evidence set belongs to another session") + persisted_evidence = self._required_artifact(record, "evidence_set") + if ( + persisted_evidence.iteration != record.session.iteration + or persisted_evidence.payload != evidence_set.to_dict() + ): + raise DomainError("evidence set is not active for this iteration") + if record.session.state == SessionState.EXECUTING: + record.session.transition(SessionState.EVALUATING) + elif record.session.state != SessionState.EVALUATING: + raise DomainError("evidence can be evaluated only after execution") + evaluation = self._evaluator.evaluate(evidence_set) + record.append("evaluation", _jsonable(evaluation)) + self._repository.save(record) + self._metrics.increment("contradictions_detected", len(evaluation.conflict_ids)) + self._emit( + "investigation.evaluated", + record, + { + "evaluation_id": evaluation.evaluation_id, + "accepted_evidence": evaluation.accepted_evidence_count, + "conflicts": len(evaluation.conflict_ids), + }, + ) + return evaluation + + def verify(self, session_id: str, evaluation: Evaluation) -> VerificationReport: + record = self._record(session_id) + if record.session.state != SessionState.EVALUATING: + raise DomainError("claim verification requires an EVALUATING session") + if evaluation.session_id != session_id: + raise DomainError("evaluation belongs to another session") + evaluation_artifact = self._required_artifact(record, "evaluation") + if ( + evaluation_artifact.iteration != record.session.iteration + or evaluation_artifact.payload != _jsonable(evaluation) + ): + raise DomainError("evaluation is not active for this iteration") + self._emit("investigation.verification_started", record, {}) + verification = self._verifier.verify(evaluation) + record.append("verification", verification.to_dict()) + self._repository.save(record) + return verification + + def update_knowledge( + self, + session_id: str, + verification: VerificationReport, + evidence_set: EvidenceSet, + ) -> tuple[InvestigationKnowledgeUpdate, KnowledgeUpdateResult]: + record = self._record(session_id) + if record.session.state != SessionState.EVALUATING: + raise DomainError("knowledge update requires an EVALUATING session") + if verification.session_id != session_id or evidence_set.session_id != session_id: + raise DomainError("verification or evidence belongs to another session") + evidence_artifact = self._required_artifact(record, "evidence_set") + if ( + evidence_artifact.iteration != record.session.iteration + or evidence_artifact.payload != evidence_set.to_dict() + ): + raise DomainError("evidence set is not active for this iteration") + verification_artifact = self._required_artifact(record, "verification") + if ( + verification_artifact.iteration != record.session.iteration + or verification_artifact.payload != verification.to_dict() + ): + raise DomainError("verification is not active for this iteration") + record.session.transition(SessionState.UPDATING) + update = self._updates.prepare(verification, evidence_set) + record.append("knowledge_update", update.to_dict()) + self._repository.save(record) + result = self._updates.apply(update) + record.append("knowledge_update_result", result.to_dict()) + self._repository.save(record) + self._metrics.increment("knowledge_updates", len(result.committed_claim_ids)) + self._emit( + "investigation.knowledge_updated", + record, + { + "update_id": update.update_id, + "committed_claim_ids": list(result.committed_claim_ids), + "unresolved_contradiction_ids": list(result.unresolved_contradiction_ids), + }, + ) + return update, result + + def finish_iteration( + self, + session_id: str, + *, + before: KnowledgeSnapshot, + evidence_set: EvidenceSet, + verification: VerificationReport, + update_result: KnowledgeUpdateResult, + objective_satisfied: bool = False, + contradictions_resolvable: bool = True, + ) -> tuple[ProgressReport, TerminationDecision]: + record = self._record(session_id) + if record.session.state != SessionState.UPDATING: + raise DomainError("iteration can finish only while UPDATING") + snapshot_artifact = self._required_artifact(record, "knowledge_snapshot") + if ( + snapshot_artifact.iteration != record.session.iteration + or snapshot_artifact.payload != before.to_dict() + ): + raise DomainError("before snapshot is not active for this iteration") + evidence_artifact = self._required_artifact(record, "evidence_set") + verification_artifact = self._required_artifact(record, "verification") + update_result_artifact = self._required_artifact(record, "knowledge_update_result") + if ( + evidence_artifact.iteration != record.session.iteration + or evidence_artifact.payload != evidence_set.to_dict() + or verification_artifact.iteration != record.session.iteration + or verification_artifact.payload != verification.to_dict() + or update_result_artifact.iteration != record.session.iteration + or update_result_artifact.payload != update_result.to_dict() + ): + raise DomainError("iteration inputs do not match active persisted artifacts") + after = self._observer.observe(record.objective) + record.append("knowledge_snapshot", after.to_dict()) + progress = ProgressMeasurer().measure( + session_id=session_id, + iteration=record.session.iteration + 1, + before_gaps=(GapState(item.id, item.uncertainty) for item in before.gaps), + after_gaps=(GapState(item.id, item.uncertainty) for item in after.gaps), + before_contradiction_ids=before.contradiction_ids, + after_contradiction_ids=after.contradiction_ids, + evidence_collected=len(evidence_set.evidence), + knowledge_updates=len(update_result.committed_claim_ids), + cost=record.session.usage.cost, + ) + record.append("progress_report", _jsonable(progress)) + record.session.complete_iteration() + confidence = ( + min(item.confidence for item in verification.decisions) + if verification.decisions and not after.gaps + else 0.0 + ) + latest_scores = record.latest("investigation_scores") + score_items = [] if latest_scores is None else latest_scores.payload.get("items", []) + best_score = None + if isinstance(score_items, list) and score_items: + best_score = max(float(item["score"]) for item in score_items if isinstance(item, dict)) + decision = self._termination.evaluate( + record.session, + TerminationContext( + objective_satisfied=objective_satisfied, + objective_confidence=confidence, + remaining_gap_count=len(after.gaps), + best_candidate_score=best_score, + unresolved_contradictions=len(update_result.unresolved_contradiction_ids), + contradictions_resolvable=contradictions_resolvable, + ), + ) + record.append("termination_decision", decision.to_dict()) + self._metrics.increment("iterations") + self._metrics.increment("gaps_resolved", len(progress.resolved_gap_ids)) + self._metrics.increment("contradictions_resolved", len(progress.contradictions_resolved)) + self._metrics.observe("average_information_gain", progress.information_gain) + if progress.cost_per_resolved_gap is not None: + self._metrics.observe("cost_per_resolved_gap", progress.cost_per_resolved_gap) + self._emit( + "investigation.iteration_completed", + record, + {"iteration": record.session.iteration, "information_gain": progress.information_gain}, + ) + self._apply_termination(record, decision) + if not decision.terminate: + self._repository.save(record) + return progress, decision + + def process_results( + self, + session_id: str, + results: Sequence[InvestigationResult], + *, + objective_satisfied: bool = False, + contradictions_resolvable: bool = True, + ) -> IterationOutcome: + """Thin composition of collection, evaluation, verification, update, and progress.""" + record = self._record(session_id) + before_artifact = self._required_artifact(record, "knowledge_snapshot") + before = KnowledgeSnapshot.from_dict(before_artifact.payload) + evidence_set = self.collect_evidence(session_id, results) + evaluation = self.evaluate(session_id, evidence_set) + verification = self.verify(session_id, evaluation) + update, update_result = self.update_knowledge(session_id, verification, evidence_set) + progress, termination = self.finish_iteration( + session_id, + before=before, + evidence_set=evidence_set, + verification=verification, + update_result=update_result, + objective_satisfied=objective_satisfied, + contradictions_resolvable=contradictions_resolvable, + ) + return IterationOutcome( + evidence_set, + evaluation, + verification, + update, + update_result, + progress, + termination, + ) + + def resume_collected_iteration( + self, + session_id: str, + *, + objective_satisfied: bool = False, + contradictions_resolvable: bool = True, + ) -> IterationOutcome: + """Resume after durable collection without rerunning agents or duplicating evidence.""" + record = self._record(session_id) + if record.session.state != SessionState.EXECUTING: + raise DomainError("collected iteration can resume only from EXECUTING") + before = KnowledgeSnapshot.from_dict( + self._required_artifact(record, "knowledge_snapshot").payload + ) + evidence_set = EvidenceSet.from_dict( + self._required_artifact(record, "evidence_set").payload + ) + evaluation = self.evaluate(session_id, evidence_set) + verification = self.verify(session_id, evaluation) + update, update_result = self.update_knowledge(session_id, verification, evidence_set) + progress, termination = self.finish_iteration( + session_id, + before=before, + evidence_set=evidence_set, + verification=verification, + update_result=update_result, + objective_satisfied=objective_satisfied, + contradictions_resolvable=contradictions_resolvable, + ) + return IterationOutcome( + evidence_set, + evaluation, + verification, + update, + update_result, + progress, + termination, + ) + + def resume_iteration( + self, + session_id: str, + *, + objective_satisfied: bool = False, + contradictions_resolvable: bool = True, + ) -> IterationOutcome: + """Resume from the latest durable stage without rerunning completed work.""" + state = self._record(session_id).session.state + if state == SessionState.EXECUTING: + return self.resume_collected_iteration( + session_id, + objective_satisfied=objective_satisfied, + contradictions_resolvable=contradictions_resolvable, + ) + if state == SessionState.EVALUATING: + return self._resume_evaluating( + session_id, + objective_satisfied=objective_satisfied, + contradictions_resolvable=contradictions_resolvable, + ) + if state == SessionState.UPDATING: + return self._resume_updating( + session_id, + objective_satisfied=objective_satisfied, + contradictions_resolvable=contradictions_resolvable, + ) + raise DomainError("session is not at a resumable iteration stage") + + def _resume_evaluating( + self, + session_id: str, + *, + objective_satisfied: bool, + contradictions_resolvable: bool, + ) -> IterationOutcome: + record = self._record(session_id) + before = KnowledgeSnapshot.from_dict( + self._required_artifact(record, "knowledge_snapshot").payload + ) + evidence_set = EvidenceSet.from_dict( + self._required_artifact(record, "evidence_set").payload + ) + verification_artifact = record.latest("verification") + if ( + verification_artifact is not None + and verification_artifact.iteration == record.session.iteration + ): + verification = VerificationReport.from_dict(verification_artifact.payload) + evaluation = self._evaluator.evaluate(evidence_set) + else: + evaluation = self.evaluate(session_id, evidence_set) + verification = self.verify(session_id, evaluation) + update, update_result = self.update_knowledge(session_id, verification, evidence_set) + progress, termination = self.finish_iteration( + session_id, + before=before, + evidence_set=evidence_set, + verification=verification, + update_result=update_result, + objective_satisfied=objective_satisfied, + contradictions_resolvable=contradictions_resolvable, + ) + return IterationOutcome( + evidence_set, + evaluation, + verification, + update, + update_result, + progress, + termination, + ) + + def _resume_updating( + self, + session_id: str, + *, + objective_satisfied: bool, + contradictions_resolvable: bool, + ) -> IterationOutcome: + record = self._record(session_id) + before = KnowledgeSnapshot.from_dict( + self._required_artifact(record, "knowledge_snapshot").payload + ) + evidence_set = EvidenceSet.from_dict( + self._required_artifact(record, "evidence_set").payload + ) + verification = VerificationReport.from_dict( + self._required_artifact(record, "verification").payload + ) + update = InvestigationKnowledgeUpdate.from_dict( + self._required_artifact(record, "knowledge_update").payload + ) + result_artifact = record.latest("knowledge_update_result") + if result_artifact is not None and result_artifact.iteration == record.session.iteration: + update_result = KnowledgeUpdateResult.from_dict(result_artifact.payload) + else: + update_result = self._updates.apply(update) + record.append("knowledge_update_result", update_result.to_dict()) + self._repository.save(record) + evaluation = self._evaluator.evaluate(evidence_set) + progress, termination = self.finish_iteration( + session_id, + before=before, + evidence_set=evidence_set, + verification=verification, + update_result=update_result, + objective_satisfied=objective_satisfied, + contradictions_resolvable=contradictions_resolvable, + ) + return IterationOutcome( + evidence_set, + evaluation, + verification, + update, + update_result, + progress, + termination, + ) + + def pause(self, session_id: str) -> InvestigationSession: + record = self._record(session_id) + record.session.pause() + assert record.session.paused_from is not None + record.append("session_paused", {"paused_from": record.session.paused_from.value}) + self._repository.save(record) + return record.session + + def resume(self, session_id: str) -> InvestigationSession: + record = self._record(session_id) + record.session.resume() + record.append("session_resumed", {"state": record.session.state.value}) + self._repository.save(record) + return record.session + + def cancel(self, session_id: str, principal: str = "user") -> InvestigationSession: + record = self._record(session_id) + execution_artifact = record.latest("plan_execution") + if execution_artifact is not None and self._execution is not None: + execution = PlanExecution.from_dict(execution_artifact.payload) + self._execution.cancel(execution, principal) + record.session.transition( + SessionState.CANCELLED, reason=TerminationReason.USER_CANCELLATION + ) + record.append("termination_decision", {"reason": TerminationReason.USER_CANCELLATION.value}) + self._repository.save(record) + self._metrics.increment("investigation_sessions_completed") + self._emit("investigation.completed", record, {"reason": "user_cancellation"}) + return record.session + + def explain(self, session_id: str) -> dict[str, object]: + record = self._record(session_id) + return { + "objective": record.objective.to_dict(), + "session": record.session.to_dict(), + "why_investigated": [ + item.payload for item in record.artifacts if item.kind == "investigation_scores" + ], + "why_believed": [ + item.payload for item in record.artifacts if item.kind == "verification" + ], + "why_stopped": [ + item.payload for item in record.artifacts if item.kind == "termination_decision" + ], + } + + def _record(self, session_id: str) -> InvestigationRecord: + record = self._repository.get(session_id) + if record is None: + raise DomainError(f"unknown investigation session: {session_id}") + return record + + @staticmethod + def _required_artifact(record: InvestigationRecord, kind: str) -> Any: + artifact = record.latest(kind) + if artifact is None: + raise DomainError(f"session has no {kind} artifact") + return artifact + + @staticmethod + def _require_snapshot(record: InvestigationRecord, snapshot: KnowledgeSnapshot) -> None: + if snapshot.objective_id != record.objective.objective_id: + raise DomainError("knowledge snapshot belongs to another objective") + if record.session.state != SessionState.PLANNING: + raise DomainError("investigation planning requires a PLANNING session") + artifact = record.latest("knowledge_snapshot") + if ( + artifact is None + or artifact.iteration != record.session.iteration + or artifact.payload != snapshot.to_dict() + ): + raise DomainError("knowledge snapshot is not active for this iteration") + + @staticmethod + def _validate_results(execution: PlanExecution, results: Sequence[InvestigationResult]) -> None: + result_ids = [item.result_id for item in results] + investigation_ids = [item.investigation_id for item in results] + if len(result_ids) != len(set(result_ids)): + raise DomainError("investigation result identifiers must be unique") + if len(investigation_ids) != len(set(investigation_ids)): + raise DomainError("only one result per investigation can be collected") + for result in results: + run_id = execution.run_ids.get(result.investigation_id) + if run_id is None or result.run_id != run_id: + raise DomainError("investigation result does not belong to the active plan") + task_id = execution.task_ids.get(result.investigation_id) + if task_id is not None and result.task_id != task_id: + raise DomainError("investigation result task does not match the active plan") + if result.investigation_id in execution.blocked_investigations: + prefix = f"blocked:{execution.plan_id}:{result.investigation_id}" + if result.task_id != prefix or result.state != InvestigationResultState.FAILED: + raise DomainError("blocked investigation result is malformed") + elif task_id is None: + raise DomainError("result was produced for an investigation that was not submitted") + + def _apply_termination( + self, record: InvestigationRecord, decision: TerminationDecision + ) -> None: + if not decision.terminate: + self._repository.save(record) + return + assert decision.target_state is not None and decision.reason is not None + record.session.transition(decision.target_state, reason=decision.reason) + self._repository.save(record) + event_type = ( + "investigation.failed" + if decision.target_state == SessionState.FAILED + else "investigation.completed" + ) + self._metrics.increment( + "investigation_sessions_failed" + if decision.target_state == SessionState.FAILED + else "investigation_sessions_completed" + ) + self._emit(event_type, record, {"reason": decision.reason.value}) + + def _emit( + self, event_type: str, record: InvestigationRecord, payload: Mapping[str, Any] + ) -> None: + self._events.publish( + Event( + event_type=event_type, + payload={ + "session_id": record.session.session_id, + "objective_id": record.objective.objective_id, + "iteration": record.session.iteration, + **dict(payload), + }, + producer="investigation-application", + trace_id=record.session.session_id, + correlation_id=record.objective.objective_id, + ) + ) + + +def _jsonable(value: Any) -> dict[str, Any]: + def default(item: object) -> object: + if isinstance(item, datetime): + return item.isoformat() + if isinstance(item, Enum): + return item.value + if is_dataclass(item): + return asdict(cast(Any, item)) + if isinstance(item, tuple | set | frozenset): + return list(item) + raise TypeError(f"cannot serialize {type(item).__name__}") + + encoded = json.dumps(value, default=default, sort_keys=True, allow_nan=False) + payload = json.loads(encoded) + if not isinstance(payload, dict): + raise DomainError("investigation artifact must serialize as an object") + return cast(dict[str, Any], payload) diff --git a/src/nexus_runtime/investigation/benchmark.py b/src/nexus_runtime/investigation/benchmark.py new file mode 100644 index 0000000..283c35b --- /dev/null +++ b/src/nexus_runtime/investigation/benchmark.py @@ -0,0 +1,180 @@ +"""Deterministic Phase 4 baseline: 10 gaps, 50 candidates, 10 tasks.""" + +from __future__ import annotations + +import json +from datetime import UTC, datetime, timedelta +from time import perf_counter + +from nexus_knowledge.domain.knowledge_gap import Investigation, KnowledgeGap +from nexus_runtime.distributed.service import RuntimeApplication +from nexus_runtime.distributed.simulator import DeterministicHarness, LocalDistributedSimulator + +from .evaluation import EvidenceEvaluator +from .evidence import ClaimStatement, Evidence, EvidenceSet +from .execution import PlanExecutionController +from .generator import InvestigationGenerator +from .objective import ResearchObjective +from .planner import InvestigationPlanner +from .provenance import EvidenceProvenance +from .scoring import InvestigationScoringModel +from .selector import InvestigationSelector +from .session import InvestigationBudget, InvestigationSession + + +def run_benchmark() -> dict[str, object]: + timestamp = datetime(2026, 1, 1, tzinfo=UTC) + objective = ResearchObjective( + question="Which ten missing facts should NEXUS resolve?", + success_criteria=("all selected gaps have evidence",), + objective_id="objective-benchmark", + created_at=timestamp, + ) + gaps = [_gap(index) for index in range(10)] + session = InvestigationSession( + objective_id=objective.objective_id, + budget=InvestigationBudget(2, 10, 10, 100.0, timedelta(minutes=10)), + session_id="session-benchmark", + created_at=timestamp, + updated_at=timestamp, + ) + + planning_started = perf_counter() + candidates = InvestigationGenerator().generate(objective, gaps) + scores = InvestigationScoringModel().score_all(candidates, gaps) + selection = InvestigationSelector().select( + scores, + budget=session.budget, + usage=session.usage, + worker_capacity=10, + top_k=10, + ) + plan = InvestigationPlanner().build(session, selection, created_at=timestamp) + planning_seconds = perf_counter() - planning_started + + simulator = LocalDistributedSimulator(start=timestamp) + harness = DeterministicHarness() + capabilities = frozenset({"document_analysis", "search"}) + for index in range(10): + simulator.add_worker(f"investigation-worker-{index}", capabilities, harness) + runtime = RuntimeApplication(simulator.coordinator) + run_ids = { + investigation.investigation_id: f"benchmark-run-{index}" + for index, investigation in enumerate(plan.investigations) + } + scheduling_started = perf_counter() + execution = PlanExecutionController(runtime).start(plan, run_ids) + scheduling_seconds = perf_counter() - scheduling_started + execution_started = perf_counter() + runtime_report = simulator.run_until_terminal() + execution_seconds = perf_counter() - execution_started + + evidence_set = EvidenceSet( + session_id=session.session_id, + evidence=tuple( + _evidence( + session.session_id, + investigation.investigation_id, + task_id, + run_ids[investigation.investigation_id], + index, + ) + for index, investigation in enumerate(plan.investigations) + for task_id in (execution.execution.task_ids[investigation.investigation_id],) + ), + ) + evaluation_started = perf_counter() + evaluation = EvidenceEvaluator().evaluate(evidence_set) + evaluation_seconds = perf_counter() - evaluation_started + total_seconds = planning_seconds + scheduling_seconds + execution_seconds + evaluation_seconds + return { + "methodology": { + "gaps": len(gaps), + "candidates": len(candidates), + "selected": len(selection.selected), + "distributed_tasks": runtime_report.tasks, + "clock": "fixed UTC inputs with local deterministic runtime", + }, + "results": { + "planning_seconds": planning_seconds, + "scheduling_seconds": scheduling_seconds, + "execution_seconds": execution_seconds, + "evidence_evaluation_seconds": evaluation_seconds, + "total_iteration_seconds": total_seconds, + "accepted_evidence": evaluation.accepted_evidence_count, + "succeeded_tasks": runtime_report.succeeded, + }, + } + + +def _gap(index: int) -> KnowledgeGap: + gap = KnowledgeGap( + id=f"gap-{index}", + kind="missing_evidence", + description=f"resolve missing fact {index}", + reason="benchmark fixture", + affected_entities=[f"entity-{index}"], + uncertainty=0.8, + importance=0.8, + estimated_cost=1.0, + created_at="2026-01-01T00:00:00Z", + ) + gap.candidate_investigations = [ + Investigation( + id=f"legacy-{index}-{candidate}", + gap_id=gap.id, + description=f"investigate fact {index} through channel {candidate}", + target_entities=[f"entity-{index}"], + estimated_cost=1.0, + metadata={"evidence_channel": f"channel-{candidate}"}, + created_at="2026-01-01T00:00:00Z", + ) + for candidate in range(5) + ] + return gap + + +def _evidence( + session_id: str, + investigation_id: str, + task_id: str, + run_id: str, + index: int, +) -> Evidence: + source_id = f"source-{index}" + return Evidence( + investigation_id=investigation_id, + source=f"benchmark://{source_id}", + claim=ClaimStatement( + text=f"fact {index} is supported", + subject=f"fact-{index}", + predicate="status", + object="supported", + claim_id=f"claim-benchmark-{index}", + ), + provenance=EvidenceProvenance( + session_id=session_id, + investigation_id=investigation_id, + task_id=task_id, + attempt_id=f"attempt-{index}", + run_id=run_id, + tool_call_id=f"tool-{index}", + source_id=source_id, + document_id=f"document-{index}", + chunk_id=f"chunk-{index}", + source_reference=f"benchmark://{source_id}", + ), + confidence=0.8, + source_quality=0.8, + excerpt=f"supporting excerpt {index}", + evidence_id=f"evidence-benchmark-{index}", + ) + + +def main() -> int: + print(json.dumps(run_benchmark(), indent=2, sort_keys=True)) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/src/nexus_runtime/investigation/cli.py b/src/nexus_runtime/investigation/cli.py new file mode 100644 index 0000000..e7a1839 --- /dev/null +++ b/src/nexus_runtime/investigation/cli.py @@ -0,0 +1,128 @@ +"""Thin command-line adapter over ``InvestigationApplication``.""" + +from __future__ import annotations + +import argparse +import json +from collections.abc import Sequence +from datetime import timedelta +from pathlib import Path + +from nexus_knowledge.service.factory import create_engine +from nexus_runtime.models import DomainError + +from .application import InvestigationApplication +from .objective import ResearchObjective +from .repository import InvestigationRecord, SQLiteInvestigationRepository +from .session import InvestigationBudget + + +def build_parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser(prog="research") + parser.add_argument("--db", type=Path, default=Path(".nexus/investigations.sqlite")) + commands = parser.add_subparsers(dest="command", required=True) + + create = commands.add_parser("create", help="create a bounded research session") + create.add_argument("question") + create.add_argument("--criterion", action="append", required=True) + create.add_argument("--scope", action="append", default=[]) + create.add_argument("--constraint", action="append", default=[]) + create.add_argument("--max-iterations", type=int, default=3) + create.add_argument("--max-investigations", type=int, default=20) + create.add_argument("--max-agent-runs", type=int, default=20) + create.add_argument("--max-cost", type=float, default=100.0) + create.add_argument("--max-execution-seconds", type=float, default=3600.0) + + for name in ( + "status", + "pause", + "resume", + "cancel", + "explain", + "evidence", + "gaps", + "plan", + "iterations", + ): + command = commands.add_parser(name) + command.add_argument("session_id") + return parser + + +def main( + argv: Sequence[str] | None = None, + *, + application: InvestigationApplication | None = None, +) -> int: + parser = build_parser() + arguments = parser.parse_args(argv) + repository: SQLiteInvestigationRepository | None = None + if application is None: + repository = SQLiteInvestigationRepository(arguments.db) + engine = create_engine() + application = InvestigationApplication(engine, repository=repository) + try: + result = _dispatch(arguments, application) + print(json.dumps(result, indent=2, sort_keys=True, default=str)) + return 0 + except (DomainError, ValueError) as exc: + parser.exit(2, f"error: {exc}\n") + finally: + if repository is not None: + repository.close() + return 2 + + +def _dispatch(arguments: argparse.Namespace, application: InvestigationApplication) -> object: + if arguments.command == "create": + objective = ResearchObjective( + question=arguments.question, + success_criteria=tuple(arguments.criterion), + scope=tuple(arguments.scope), + constraints=tuple(arguments.constraint), + ) + budget = InvestigationBudget( + max_iterations=arguments.max_iterations, + max_investigations=arguments.max_investigations, + max_agent_runs=arguments.max_agent_runs, + max_cost=arguments.max_cost, + max_execution_time=timedelta(seconds=arguments.max_execution_seconds), + ) + return application.create(objective, budget).to_dict() + session_id = str(arguments.session_id) + if arguments.command == "status": + return application.status(session_id).to_dict() + if arguments.command == "pause": + return application.pause(session_id).to_dict() + if arguments.command == "resume": + return application.resume(session_id).to_dict() + if arguments.command == "cancel": + return application.cancel(session_id).to_dict() + if arguments.command == "explain": + return application.explain(session_id) + record = application.status(session_id) + if arguments.command == "evidence": + return _artifacts(record, "evidence_set") + if arguments.command == "gaps": + snapshots = _artifacts(record, "knowledge_snapshot") + return [] if not snapshots else snapshots[-1].get("gaps", []) + if arguments.command == "plan": + plans = _artifacts(record, "investigation_plan") + return None if not plans else plans[-1] + if arguments.command == "iterations": + return [ + { + "iteration": iteration, + "artifacts": [item.to_dict() for item in record.for_iteration(iteration)], + } + for iteration in sorted({item.iteration for item in record.artifacts}) + ] + raise DomainError(f"unknown research command: {arguments.command}") + + +def _artifacts(record: InvestigationRecord, kind: str) -> list[dict[str, object]]: + return [dict(item.payload) for item in record.artifacts if item.kind == kind] + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/src/nexus_runtime/investigation/evidence.py b/src/nexus_runtime/investigation/evidence.py index 941633e..6985d92 100644 --- a/src/nexus_runtime/investigation/evidence.py +++ b/src/nexus_runtime/investigation/evidence.py @@ -4,6 +4,7 @@ import hashlib import json +from collections.abc import Mapping from dataclasses import dataclass, field from datetime import datetime from enum import StrEnum @@ -19,6 +20,34 @@ def _required(value: str, name: str) -> None: raise ValueError(f"{name} must be a non-empty string") +def _persisted_string(payload: Mapping[str, Any], name: str) -> str: + value = payload.get(name) + if not isinstance(value, str): + raise ValueError(f"{name} must be a string") + return value + + +def _persisted_float(payload: Mapping[str, Any], name: str) -> float: + value = payload.get(name) + if isinstance(value, bool) or not isinstance(value, (int, float)): + raise ValueError(f"{name} must be a number") + return float(value) + + +def _persisted_bool(payload: Mapping[str, Any], name: str) -> bool: + value = payload.get(name) + if not isinstance(value, bool): + raise ValueError(f"{name} must be a boolean") + return value + + +def _persisted_strings(payload: Mapping[str, Any], name: str) -> tuple[str, ...]: + value = payload.get(name) + if not isinstance(value, list) or any(not isinstance(item, str) for item in value): + raise ValueError(f"{name} must be a list of strings") + return tuple(value) + + class EvidenceRole(StrEnum): """How an evidence item relates to its asserted claim.""" @@ -75,6 +104,19 @@ def to_dict(self) -> dict[str, str]: "object": self.object, } + @classmethod + def from_dict(cls, payload: Mapping[str, Any]) -> ClaimStatement: + try: + return cls( + claim_id=_persisted_string(payload, "claim_id"), + text=_persisted_string(payload, "text"), + subject=_persisted_string(payload, "subject"), + predicate=_persisted_string(payload, "predicate"), + object=_persisted_string(payload, "object"), + ) + except KeyError as exc: + raise ValueError("malformed claim statement") from exc + @dataclass(frozen=True, slots=True) class Evidence: @@ -144,6 +186,44 @@ def to_dict(self) -> dict[str, Any]: "metadata": self.metadata, } + @classmethod + def from_dict(cls, payload: Mapping[str, object]) -> Evidence: + claim = payload.get("claim") + provenance = payload.get("provenance") + item_payload = payload.get("payload", {}) + metadata = payload.get("metadata", {}) + entities = payload.get("supporting_entities", []) + if ( + not isinstance(claim, dict) + or not isinstance(provenance, dict) + or not isinstance(item_payload, dict) + or not isinstance(metadata, dict) + or not isinstance(entities, list) + ): + raise ValueError("malformed evidence") + excerpt = payload.get("excerpt", "") + error_role = payload.get("role") + if not isinstance(excerpt, str) or not isinstance(error_role, str): + raise ValueError("malformed evidence") + try: + return cls( + evidence_id=_persisted_string(payload, "evidence_id"), + investigation_id=_persisted_string(payload, "investigation_id"), + source=_persisted_string(payload, "source"), + claim=ClaimStatement.from_dict(claim), + excerpt=excerpt, + payload=dict(item_payload), + provenance=EvidenceProvenance.from_dict(provenance), + confidence=_persisted_float(payload, "confidence"), + source_quality=_persisted_float(payload, "source_quality"), + role=EvidenceRole(error_role), + timestamp=_parse_timestamp(payload["timestamp"]), + supporting_entities=_persisted_strings(payload, "supporting_entities"), + metadata=dict(metadata), + ) + except (KeyError, TypeError, ValueError) as exc: + raise ValueError("malformed evidence") from exc + @dataclass(frozen=True, slots=True) class EvidenceSet: @@ -167,6 +247,11 @@ def __post_init__(self) -> None: ids = [item.evidence_id for item in self.evidence] if len(ids) != len(set(ids)): raise ValueError("evidence_id values must be unique within an EvidenceSet") + identities_by_claim_id: dict[str, tuple[str, str, str]] = {} + for item in self.evidence: + previous = identities_by_claim_id.setdefault(item.claim.claim_id, item.claim.identity) + if previous != item.claim.identity: + raise ValueError("a claim_id cannot represent multiple claim identities") @property def investigation_ids(self) -> tuple[str, ...]: @@ -180,6 +265,21 @@ def to_dict(self) -> dict[str, Any]: "evidence": [item.to_dict() for item in self.evidence], } + @classmethod + def from_dict(cls, payload: Mapping[str, object]) -> EvidenceSet: + evidence = payload.get("evidence") + if not isinstance(evidence, list) or any(not isinstance(item, dict) for item in evidence): + raise ValueError("malformed evidence set") + try: + return cls( + evidence_set_id=_persisted_string(payload, "evidence_set_id"), + session_id=_persisted_string(payload, "session_id"), + created_at=_parse_timestamp(payload["created_at"]), + evidence=tuple(Evidence.from_dict(item) for item in evidence), + ) + except (KeyError, TypeError, ValueError) as exc: + raise ValueError("malformed evidence set") from exc + @dataclass(frozen=True, slots=True) class InvestigationResult: @@ -244,3 +344,39 @@ def to_dict(self) -> dict[str, Any]: "metadata": self.metadata, "completed_at": self.completed_at.isoformat(), } + + @classmethod + def from_dict(cls, payload: Mapping[str, object]) -> InvestigationResult: + evidence_set = payload.get("evidence_set") + metadata = payload.get("metadata", {}) + if not isinstance(evidence_set, dict) or not isinstance(metadata, dict): + raise ValueError("malformed investigation result") + error = payload.get("error") + state = payload.get("state") + if (error is not None and not isinstance(error, str)) or not isinstance(state, str): + raise ValueError("malformed investigation result") + try: + return cls( + result_id=_persisted_string(payload, "result_id"), + session_id=_persisted_string(payload, "session_id"), + investigation_id=_persisted_string(payload, "investigation_id"), + task_id=_persisted_string(payload, "task_id"), + attempt_id=_persisted_string(payload, "attempt_id"), + run_id=_persisted_string(payload, "run_id"), + state=InvestigationResultState(state), + evidence_set=EvidenceSet.from_dict(evidence_set), + error=error, + metadata=dict(metadata), + completed_at=_parse_timestamp(payload["completed_at"]), + ) + except (KeyError, TypeError, ValueError) as exc: + raise ValueError("malformed investigation result") from exc + + +def _parse_timestamp(value: object) -> datetime: + if not isinstance(value, str): + raise ValueError("timestamp must be an ISO-8601 string") + try: + return datetime.fromisoformat(value.replace("Z", "+00:00")) + except ValueError as exc: + raise ValueError("timestamp must be an ISO-8601 string") from exc diff --git a/src/nexus_runtime/investigation/execution.py b/src/nexus_runtime/investigation/execution.py new file mode 100644 index 0000000..88b399f --- /dev/null +++ b/src/nexus_runtime/investigation/execution.py @@ -0,0 +1,281 @@ +"""Submit investigation plans to the existing distributed runtime in DAG-ready waves.""" + +from __future__ import annotations + +from collections.abc import Mapping +from dataclasses import dataclass, field +from datetime import datetime +from typing import Any, Protocol + +from nexus_runtime.distributed.model import ( + TERMINAL_TASK_STATES, + DistributedTask, + DistributedTaskState, + TaskPriority, +) +from nexus_runtime.models import DomainError, utcnow + +from .objective import _required_string, _timestamp_from_text, _timestamp_to_text +from .planner import InvestigationPlan + + +class DistributedRuntimePort(Protocol): + """Public subset of ``RuntimeApplication`` used by the investigation layer.""" + + def submit_task( + self, + run_id: str, + *, + correlation_id: str, + priority: TaskPriority = TaskPriority.NORMAL, + required_capabilities: frozenset[str] = frozenset(), + metadata: dict[str, Any] | None = None, + ) -> DistributedTask: ... + + def get_task(self, task_id: str) -> DistributedTask: ... + + def cancel_task(self, task_id: str, principal: str) -> DistributedTask: ... + + def list_tasks(self) -> list[DistributedTask]: ... + + +@dataclass(slots=True) +class PlanExecution: + """Persistable bridge between a plan, AgentRuns, and distributed task ids.""" + + plan_id: str + session_id: str + run_ids: dict[str, str] + task_ids: dict[str, str] = field(default_factory=dict) + blocked_investigations: dict[str, str] = field(default_factory=dict) + created_at: datetime = field(default_factory=utcnow) + updated_at: datetime = field(default_factory=utcnow) + + def __post_init__(self) -> None: + if not self.plan_id.strip() or not self.session_id.strip(): + raise DomainError("execution plan_id and session_id are required") + if any(not key.strip() or not value.strip() for key, value in self.run_ids.items()): + raise DomainError("execution AgentRun mapping cannot contain empty identifiers") + + @property + def all_scheduled_or_blocked(self) -> bool: + return len(self.task_ids) + len(self.blocked_investigations) == len(self.run_ids) + + def to_dict(self) -> dict[str, object]: + return { + "plan_id": self.plan_id, + "session_id": self.session_id, + "run_ids": dict(self.run_ids), + "task_ids": dict(self.task_ids), + "blocked_investigations": dict(self.blocked_investigations), + "created_at": _timestamp_to_text(self.created_at), + "updated_at": _timestamp_to_text(self.updated_at), + } + + @classmethod + def from_dict(cls, payload: Mapping[str, object]) -> PlanExecution: + run_ids = payload.get("run_ids") + task_ids = payload.get("task_ids") + blocked = payload.get("blocked_investigations") + if ( + not isinstance(run_ids, dict) + or not isinstance(task_ids, dict) + or not isinstance(blocked, dict) + ): + raise DomainError("malformed PlanExecution") + for name, value in ( + ("run_ids", run_ids), + ("task_ids", task_ids), + ("blocked_investigations", blocked), + ): + if any( + not isinstance(key, str) or not isinstance(item, str) for key, item in value.items() + ): + raise DomainError(f"malformed PlanExecution {name}") + try: + return cls( + plan_id=_required_string(payload["plan_id"], "plan_id"), + session_id=_required_string(payload["session_id"], "session_id"), + run_ids=dict(run_ids), + task_ids=dict(task_ids), + blocked_investigations=dict(blocked), + created_at=_timestamp_from_text(payload["created_at"], "created_at"), + updated_at=_timestamp_from_text(payload["updated_at"], "updated_at"), + ) + except KeyError as exc: + raise DomainError("malformed PlanExecution") from exc + + +@dataclass(frozen=True, slots=True) +class ExecutionStatus: + execution: PlanExecution + task_states: dict[str, DistributedTaskState] + terminal: bool + succeeded: tuple[str, ...] + failed: tuple[str, ...] + running: tuple[str, ...] + + +class PlanExecutionController: + """Respect plan dependencies without duplicating runtime scheduling semantics.""" + + def __init__(self, runtime: DistributedRuntimePort) -> None: + self._runtime = runtime + + def start( + self, + plan: InvestigationPlan, + run_ids: Mapping[str, str], + ) -> ExecutionStatus: + return self.advance(plan, self.prepare(plan, run_ids)) + + def prepare( + self, + plan: InvestigationPlan, + run_ids: Mapping[str, str], + ) -> PlanExecution: + expected = set(plan.dependencies) + if set(run_ids) != expected: + missing = sorted(expected - set(run_ids)) + extra = sorted(set(run_ids) - expected) + raise DomainError(f"AgentRun mapping mismatch; missing={missing}, extra={extra}") + return PlanExecution( + plan_id=plan.plan_id, + session_id=plan.session_id, + run_ids=dict(run_ids), + created_at=plan.created_at, + updated_at=plan.created_at, + ) + + def advance( + self, + plan: InvestigationPlan, + execution: PlanExecution, + ) -> ExecutionStatus: + self._validate(plan, execution) + states = self._states(execution) + made_progress = True + while made_progress: + made_progress = False + for investigation_id in sorted(plan.dependencies): + if ( + investigation_id in execution.task_ids + or investigation_id in execution.blocked_investigations + ): + continue + parents = plan.dependencies[investigation_id] + failed_parents = [ + parent + for parent in parents + if parent in execution.blocked_investigations + or states.get(parent) + in {DistributedTaskState.CANCELLED, DistributedTaskState.DEAD_LETTERED} + ] + if failed_parents: + execution.blocked_investigations[investigation_id] = ( + "dependency failed: " + ", ".join(sorted(failed_parents)) + ) + made_progress = True + continue + if not all( + states.get(parent) == DistributedTaskState.SUCCEEDED for parent in parents + ): + continue + task = self._submit_or_recover( + plan, investigation_id, execution.run_ids[investigation_id] + ) + execution.task_ids[investigation_id] = task.task_id + states[investigation_id] = task.state + made_progress = True + execution.updated_at = utcnow() + states = self._states(execution) + succeeded = tuple( + sorted(key for key, value in states.items() if value == DistributedTaskState.SUCCEEDED) + ) + failed = tuple( + sorted( + set(execution.blocked_investigations) + | { + key + for key, value in states.items() + if value in {DistributedTaskState.CANCELLED, DistributedTaskState.DEAD_LETTERED} + } + ) + ) + running = tuple( + sorted(key for key, value in states.items() if value not in TERMINAL_TASK_STATES) + ) + terminal = len(succeeded) + len(failed) == len(plan.investigations) + return ExecutionStatus(execution, states, terminal, succeeded, failed, running) + + def cancel(self, execution: PlanExecution, principal: str) -> PlanExecution: + for task_id in execution.task_ids.values(): + task = self._runtime.get_task(task_id) + if task.state not in TERMINAL_TASK_STATES: + self._runtime.cancel_task(task_id, principal) + execution.updated_at = utcnow() + return execution + + def tasks(self, execution: PlanExecution) -> tuple[DistributedTask, ...]: + return tuple(self._runtime.get_task(task_id) for task_id in execution.task_ids.values()) + + def _submit_or_recover( + self, + plan: InvestigationPlan, + investigation_id: str, + run_id: str, + ) -> DistributedTask: + existing = [ + task + for task in self._runtime.list_tasks() + if task.run_id == run_id + and task.correlation_id == plan.session_id + and task.metadata.get("plan_id") == plan.plan_id + and task.metadata.get("investigation_id") == investigation_id + ] + if len(existing) > 1: + raise DomainError(f"duplicate distributed tasks for investigation: {investigation_id}") + if existing: + return existing[0] + investigation = next( + item for item in plan.investigations if item.investigation_id == investigation_id + ) + metadata: dict[str, Any] = { + "plan_id": plan.plan_id, + "session_id": plan.session_id, + "investigation_id": investigation_id, + "gap_id": investigation.gap_id, + "question": investigation.question, + "hypothesis": investigation.hypothesis, + "required_evidence": list(investigation.required_evidence), + "constraints": list(investigation.constraints), + "dependency_investigation_ids": list(plan.dependencies[investigation_id]), + } + return self._runtime.submit_task( + run_id, + correlation_id=plan.session_id, + priority=self._priority(investigation.priority), + required_capabilities=frozenset(investigation.capabilities), + metadata=metadata, + ) + + def _states(self, execution: PlanExecution) -> dict[str, DistributedTaskState]: + return { + investigation_id: self._runtime.get_task(task_id).state + for investigation_id, task_id in execution.task_ids.items() + } + + @staticmethod + def _validate(plan: InvestigationPlan, execution: PlanExecution) -> None: + if execution.plan_id != plan.plan_id or execution.session_id != plan.session_id: + raise DomainError("execution does not belong to the investigation plan") + if set(execution.run_ids) != set(plan.dependencies): + raise DomainError("execution AgentRun mapping does not match the plan") + + @staticmethod + def _priority(value: float) -> TaskPriority: + if value >= 0.67: + return TaskPriority.HIGH + if value <= 0.33: + return TaskPriority.LOW + return TaskPriority.NORMAL diff --git a/src/nexus_runtime/investigation/generator.py b/src/nexus_runtime/investigation/generator.py index e3c1fd0..f60f363 100644 --- a/src/nexus_runtime/investigation/generator.py +++ b/src/nexus_runtime/investigation/generator.py @@ -12,6 +12,7 @@ from .objective import ( ResearchObjective, + _required_string, _string_tuple, _timestamp_from_text, _timestamp_to_text, @@ -21,7 +22,7 @@ def _as_float(value: object) -> float: - if isinstance(value, bool) or not isinstance(value, (int, float, str)): + if isinstance(value, bool) or not isinstance(value, (int, float)): raise DomainError("numeric candidate field is malformed") return float(value) @@ -217,10 +218,10 @@ def from_dict(cls, payload: Mapping[str, object]) -> CandidateInvestigation: raise DomainError("candidate metadata must be an object") try: return cls( - investigation_id=str(payload["investigation_id"]), - gap_id=str(payload["gap_id"]), - question=str(payload["question"]), - hypothesis=str(payload["hypothesis"]), + investigation_id=_required_string(payload["investigation_id"], "investigation_id"), + gap_id=_required_string(payload["gap_id"], "gap_id"), + question=_required_string(payload["question"], "question"), + hypothesis=_required_string(payload["hypothesis"], "hypothesis"), required_evidence=_string_tuple( payload["required_evidence"], "required_evidence", allow_empty=False ), diff --git a/src/nexus_runtime/investigation/knowledge_update.py b/src/nexus_runtime/investigation/knowledge_update.py index 27fd5a8..a8e4ad9 100644 --- a/src/nexus_runtime/investigation/knowledge_update.py +++ b/src/nexus_runtime/investigation/knowledge_update.py @@ -5,7 +5,7 @@ import json from dataclasses import dataclass, field from datetime import datetime -from typing import Protocol +from typing import Any, Protocol from nexus_knowledge.domain.claim import ( Claim, @@ -22,7 +22,13 @@ from nexus_knowledge.service.engine import KnowledgeUpdate, KnowledgeUpdateReceipt from nexus_runtime.models import new_id, utcnow -from .evidence import Evidence, EvidenceSet +from .evidence import ( + Evidence, + EvidenceSet, + _parse_timestamp, + _persisted_string, + _persisted_strings, +) from .provenance import EvidenceProvenance from .verification import VerificationDecision, VerificationReport @@ -36,6 +42,14 @@ def detect_contradictions(self) -> list[Contradiction]: ... def verify_claim(self, claim_id: str) -> UncertaintyAssessment: ... + def validate_evidence_provenance( + self, + source_id: str, + document_id: str, + chunk_id: str, + source_reference: str, + ) -> bool: ... + @dataclass(frozen=True, slots=True) class PreparedClaim: @@ -50,6 +64,25 @@ def __post_init__(self) -> None: if not self.decision.eligible_for_update: raise ValueError("ineligible claim cannot be prepared for knowledge update") + def to_dict(self) -> dict[str, Any]: + return { + "decision": self.decision.to_dict(), + "evidence": [item.to_dict() for item in self.evidence], + } + + @classmethod + def from_dict(cls, payload: dict[str, Any]) -> PreparedClaim: + decision = payload.get("decision") + evidence = payload.get("evidence") + if not isinstance(decision, dict) or not isinstance(evidence, list): + raise ValueError("malformed prepared claim") + if any(not isinstance(item, dict) for item in evidence): + raise ValueError("malformed prepared claim evidence") + return cls( + decision=VerificationDecision.from_dict(decision), + evidence=tuple(Evidence.from_dict(item) for item in evidence), + ) + @dataclass(frozen=True, slots=True) class InvestigationKnowledgeUpdate: @@ -69,6 +102,31 @@ def evidence_ids(self) -> tuple[str, ...]: def provenance(self) -> tuple[EvidenceProvenance, ...]: return tuple(item.provenance for claim in self.claims for item in claim.evidence) + def to_dict(self) -> dict[str, Any]: + return { + "update_id": self.update_id, + "session_id": self.session_id, + "verification_id": self.verification_id, + "claims": [item.to_dict() for item in self.claims], + "created_at": self.created_at.isoformat(), + } + + @classmethod + def from_dict(cls, payload: dict[str, Any]) -> InvestigationKnowledgeUpdate: + claims = payload.get("claims") + if not isinstance(claims, list) or any(not isinstance(item, dict) for item in claims): + raise ValueError("malformed investigation knowledge update") + try: + return cls( + update_id=_persisted_string(payload, "update_id"), + session_id=_persisted_string(payload, "session_id"), + verification_id=_persisted_string(payload, "verification_id"), + claims=tuple(PreparedClaim.from_dict(item) for item in claims), + created_at=_parse_timestamp(payload["created_at"]), + ) + except (KeyError, TypeError, ValueError) as exc: + raise ValueError("malformed investigation knowledge update") from exc + @dataclass(frozen=True, slots=True) class KnowledgeUpdateResult: @@ -85,6 +143,58 @@ class KnowledgeUpdateResult: def fully_applied(self) -> bool: return self.rejected_records == 0 and not self.unresolved_contradiction_ids + def to_dict(self) -> dict[str, Any]: + return { + "update_id": self.update_id, + "accepted_records": self.accepted_records, + "rejected_records": self.rejected_records, + "errors": list(self.errors), + "committed_claim_ids": list(self.committed_claim_ids), + "unresolved_contradiction_ids": list(self.unresolved_contradiction_ids), + "verification_states": dict(self.verification_states), + "applied_at": self.applied_at.isoformat(), + } + + @classmethod + def from_dict(cls, payload: dict[str, Any]) -> KnowledgeUpdateResult: + sequence_fields = ( + "errors", + "committed_claim_ids", + "unresolved_contradiction_ids", + ) + states = payload.get("verification_states") + invalid_sequences = any(not isinstance(payload.get(name), list) for name in sequence_fields) + if invalid_sequences or not isinstance(states, dict): + raise ValueError("malformed knowledge update result") + accepted = payload.get("accepted_records") + rejected = payload.get("rejected_records") + if ( + isinstance(accepted, bool) + or not isinstance(accepted, int) + or isinstance(rejected, bool) + or not isinstance(rejected, int) + ): + raise ValueError("malformed knowledge update result") + if any( + not isinstance(key, str) or not isinstance(value, str) for key, value in states.items() + ): + raise ValueError("malformed knowledge update result") + try: + return cls( + update_id=_persisted_string(payload, "update_id"), + accepted_records=accepted, + rejected_records=rejected, + errors=_persisted_strings(payload, "errors"), + committed_claim_ids=_persisted_strings(payload, "committed_claim_ids"), + unresolved_contradiction_ids=_persisted_strings( + payload, "unresolved_contradiction_ids" + ), + verification_states=dict(states), + applied_at=_parse_timestamp(payload["applied_at"]), + ) + except (KeyError, TypeError, ValueError) as exc: + raise ValueError("malformed knowledge update result") from exc + class KnowledgeUpdateIntegrator: """Submit only verified claims through ``commit_knowledge_update``. @@ -119,6 +229,20 @@ def prepare( evidence = tuple(evidence_by_id[item] for item in decision.supporting_evidence_ids) if any(not item.provenance.is_complete for item in evidence): raise ValueError("knowledge update requires complete evidence provenance") + invalid = [ + item.evidence_id + for item in evidence + if not self._knowledge.validate_evidence_provenance( + item.provenance.source_id, + item.provenance.document_id, + item.provenance.chunk_id, + item.provenance.source_reference, + ) + ] + if invalid: + raise ValueError( + "knowledge update provenance does not resolve: " + ", ".join(invalid) + ) prepared.append(PreparedClaim(decision=decision, evidence=evidence)) return InvestigationKnowledgeUpdate( session_id=report.session_id, diff --git a/src/nexus_runtime/investigation/metrics.py b/src/nexus_runtime/investigation/metrics.py new file mode 100644 index 0000000..a0822bd --- /dev/null +++ b/src/nexus_runtime/investigation/metrics.py @@ -0,0 +1,52 @@ +"""Scientific observability for autonomous investigation sessions.""" + +from __future__ import annotations + +from collections import defaultdict +from threading import RLock +from typing import Protocol + + +class InvestigationMetrics(Protocol): + def increment(self, name: str, amount: int = 1) -> None: ... + + def observe(self, name: str, value: float) -> None: ... + + +class InMemoryInvestigationMetrics: + """Small deterministic sink suitable for adapters and test baselines.""" + + def __init__(self) -> None: + self._counters: dict[str, int] = defaultdict(int) + self._observations: dict[str, list[float]] = defaultdict(list) + self._lock = RLock() + + def increment(self, name: str, amount: int = 1) -> None: + if amount < 0: + raise ValueError("metric increments cannot be negative") + with self._lock: + self._counters[name] += amount + + def observe(self, name: str, value: float) -> None: + with self._lock: + self._observations[name].append(float(value)) + + def snapshot(self) -> dict[str, object]: + with self._lock: + counters = dict(sorted(self._counters.items())) + observations = { + name: { + "count": len(values), + "sum": sum(values), + "average": sum(values) / len(values) if values else 0.0, + "minimum": min(values) if values else 0.0, + "maximum": max(values) if values else 0.0, + } + for name, values in sorted(self._observations.items()) + } + executed = counters.get("investigations_executed", 0) + succeeded = counters.get("investigation_successes", 0) + gauges = { + "investigation_success_rate": succeeded / executed if executed else 0.0, + } + return {"counters": counters, "observations": observations, "gauges": gauges} diff --git a/src/nexus_runtime/investigation/objective.py b/src/nexus_runtime/investigation/objective.py index 1d59a74..5827177 100644 --- a/src/nexus_runtime/investigation/objective.py +++ b/src/nexus_runtime/investigation/objective.py @@ -38,10 +38,18 @@ def _validate_json(value: object, field_name: str) -> None: raise DomainError(f"{field_name} must be JSON serializable") from exc +def _required_string(value: object, field_name: str) -> str: + if not isinstance(value, str): + raise DomainError(f"{field_name} must be a string") + return value + + def _string_tuple(value: object, field_name: str, *, allow_empty: bool = True) -> tuple[str, ...]: if not isinstance(value, (list, tuple)): raise DomainError(f"{field_name} must be a sequence of strings") - result = tuple(str(item).strip() for item in value) + if any(not isinstance(item, str) for item in value): + raise DomainError(f"{field_name} must be a sequence of strings") + result = tuple(item.strip() for item in value) if any(not item for item in result): raise DomainError(f"{field_name} cannot contain empty values") if not allow_empty and not result: @@ -111,8 +119,8 @@ def from_dict(cls, payload: Mapping[str, object]) -> ResearchObjective: if not isinstance(metadata, dict): raise DomainError("objective metadata must be an object") return cls( - objective_id=str(payload["objective_id"]), - question=str(payload["question"]), + objective_id=_required_string(payload["objective_id"], "objective_id"), + question=_required_string(payload["question"], "question"), scope=_string_tuple(payload["scope"], "scope"), constraints=_string_tuple(payload["constraints"], "constraints"), success_criteria=_string_tuple( diff --git a/src/nexus_runtime/investigation/observation.py b/src/nexus_runtime/investigation/observation.py index 25cef8a..16ad327 100644 --- a/src/nexus_runtime/investigation/observation.py +++ b/src/nexus_runtime/investigation/observation.py @@ -13,6 +13,7 @@ from .generator import ExistingInvestigation, KnowledgeGapLike, _stable_id from .objective import ( ResearchObjective, + _required_string, _string_tuple, _timestamp_from_text, _timestamp_to_text, @@ -22,7 +23,7 @@ def _number(value: object, field_name: str) -> float: - if isinstance(value, bool) or not isinstance(value, (int, float, str)): + if isinstance(value, bool) or not isinstance(value, (int, float)): raise DomainError(f"{field_name} must be numeric") return float(value) @@ -83,9 +84,9 @@ def _deserialize_gap(payload: object) -> KnowledgeGapLike: raise DomainError("candidate investigation must be an object") candidates.append( investigation_type( - id=str(raw["id"]), - gap_id=str(raw["gap_id"]), - description=str(raw["description"]), + id=_required_string(raw["id"], "candidate id"), + gap_id=_required_string(raw["gap_id"], "candidate gap_id"), + description=_required_string(raw["description"], "candidate description"), target_entities=list( _string_tuple(raw.get("target_entities", []), "target_entities") ), @@ -100,15 +101,15 @@ def _deserialize_gap(payload: object) -> KnowledgeGapLike: estimated_cost=_number(raw.get("estimated_cost", 0.0), "estimated_cost"), score=_number(raw.get("score", 0.0), "score"), metadata=dict(raw.get("metadata", {})), - created_at=str(raw["created_at"]), + created_at=_required_string(raw["created_at"], "candidate created_at"), ) ) try: result = gap_type( - id=str(payload["id"]), - kind=str(payload["kind"]), - description=str(payload["description"]), - reason=str(payload["reason"]), + id=_required_string(payload["id"], "gap id"), + kind=_required_string(payload["kind"], "gap kind"), + description=_required_string(payload["description"], "gap description"), + reason=_required_string(payload["reason"], "gap reason"), affected_entities=list( _string_tuple(payload.get("affected_entities", []), "affected_entities") ), @@ -123,7 +124,7 @@ def _deserialize_gap(payload: object) -> KnowledgeGapLike: estimated_cost=_number(payload.get("estimated_cost", 0.0), "estimated_cost"), candidate_investigations=candidates, metadata=dict(payload.get("metadata", {})), - created_at=str(payload["created_at"]), + created_at=_required_string(payload["created_at"], "gap created_at"), ) except KeyError as exc: raise DomainError("malformed knowledge snapshot gap") from exc @@ -236,16 +237,16 @@ def from_dict(cls, payload: Mapping[str, object]) -> KnowledgeSnapshot: raise DomainError("malformed KnowledgeSnapshot") try: return cls( - snapshot_id=str(payload["snapshot_id"]), - objective_id=str(payload["objective_id"]), - query=str(payload["query"]), + snapshot_id=_required_string(payload["snapshot_id"], "snapshot_id"), + objective_id=_required_string(payload["objective_id"], "objective_id"), + query=_required_string(payload["query"], "query"), gaps=tuple(_deserialize_gap(gap) for gap in raw_gaps), retrieval_refs=_string_tuple(payload["retrieval_refs"], "retrieval_refs"), entity_ids=_string_tuple(payload["entity_ids"], "entity_ids"), relation_ids=_string_tuple(payload["relation_ids"], "relation_ids"), contradiction_ids=_string_tuple(payload["contradiction_ids"], "contradiction_ids"), mean_uncertainty=_number(payload["mean_uncertainty"], "mean_uncertainty"), - summary=str(payload["summary"]), + summary=_required_string(payload["summary"], "summary"), metadata=dict(metadata), observed_at=_timestamp_from_text(payload["observed_at"], "observed_at"), ) diff --git a/src/nexus_runtime/investigation/planner.py b/src/nexus_runtime/investigation/planner.py index 2df0df3..38103d2 100644 --- a/src/nexus_runtime/investigation/planner.py +++ b/src/nexus_runtime/investigation/planner.py @@ -11,7 +11,12 @@ from nexus_runtime.models import DomainError, ResourceRequirements, Task from .generator import CandidateInvestigation, _stable_id -from .objective import _timestamp_from_text, _timestamp_to_text, _validate_timestamp +from .objective import ( + _required_string, + _timestamp_from_text, + _timestamp_to_text, + _validate_timestamp, +) from .selector import SelectionResult from .session import InvestigationBudget, InvestigationSession, SessionState @@ -156,13 +161,17 @@ def from_dict(cls, payload: Mapping[str, object]) -> InvestigationPlan: raise DomainError("malformed InvestigationPlan") normalized_dependencies: dict[str, tuple[str, ...]] = {} for key, value in dependencies.items(): - if not isinstance(key, str) or not isinstance(value, list): + if ( + not isinstance(key, str) + or not isinstance(value, list) + or any(not isinstance(item, str) for item in value) + ): raise DomainError("malformed plan dependencies") - normalized_dependencies[key] = tuple(str(item) for item in value) + normalized_dependencies[key] = tuple(value) try: return cls( - plan_id=str(payload["plan_id"]), - session_id=str(payload["session_id"]), + plan_id=_required_string(payload["plan_id"], "plan_id"), + session_id=_required_string(payload["session_id"], "session_id"), investigations=tuple( CandidateInvestigation.from_dict(item) for item in investigations diff --git a/src/nexus_runtime/investigation/provenance.py b/src/nexus_runtime/investigation/provenance.py index 3ea3f8e..74c7d6e 100644 --- a/src/nexus_runtime/investigation/provenance.py +++ b/src/nexus_runtime/investigation/provenance.py @@ -7,6 +7,7 @@ from __future__ import annotations +from collections.abc import Mapping from dataclasses import dataclass _LINEAGE_FIELDS = ( @@ -82,3 +83,14 @@ def to_dict(self) -> dict[str, str]: "chunk_id": self.chunk_id, "source_reference": self.source_reference, } + + @classmethod + def from_dict(cls, payload: Mapping[str, object]) -> EvidenceProvenance: + fields = (*_LINEAGE_FIELDS, "source_reference") + values: dict[str, str] = {} + for name in fields: + value = payload.get(name) + if not isinstance(value, str): + raise ValueError(f"malformed evidence provenance field: {name}") + values[name] = value + return cls(**values) diff --git a/src/nexus_runtime/investigation/repository.py b/src/nexus_runtime/investigation/repository.py new file mode 100644 index 0000000..48b3f36 --- /dev/null +++ b/src/nexus_runtime/investigation/repository.py @@ -0,0 +1,224 @@ +"""Durable, provider-neutral records for reconstructing investigation sessions.""" + +from __future__ import annotations + +import json +import sqlite3 +from collections.abc import Mapping +from dataclasses import dataclass, field +from datetime import datetime +from pathlib import Path +from threading import RLock +from typing import Any, Protocol + +from nexus_runtime.models import DomainError, new_id, utcnow + +from .objective import ( + ResearchObjective, + _required_string, + _timestamp_from_text, + _timestamp_to_text, +) +from .session import InvestigationSession + + +@dataclass(frozen=True, slots=True) +class InvestigationArtifact: + """One immutable, serializable decision or observation in a session.""" + + kind: str + iteration: int + payload: dict[str, Any] + artifact_id: str = field(default_factory=lambda: new_id("investigation_artifact")) + created_at: datetime = field(default_factory=utcnow) + + def __post_init__(self) -> None: + if not self.kind.strip() or not self.artifact_id.strip(): + raise DomainError("artifact kind and identifier are required") + if self.iteration < 0: + raise DomainError("artifact iteration cannot be negative") + try: + json.dumps(self.payload, sort_keys=True, allow_nan=False) + except (TypeError, ValueError) as exc: + raise DomainError("artifact payload must be JSON serializable") from exc + + def to_dict(self) -> dict[str, object]: + return { + "artifact_id": self.artifact_id, + "kind": self.kind, + "iteration": self.iteration, + "payload": dict(self.payload), + "created_at": _timestamp_to_text(self.created_at), + } + + @classmethod + def from_dict(cls, payload: Mapping[str, object]) -> InvestigationArtifact: + body = payload.get("payload") + if not isinstance(body, dict): + raise DomainError("malformed InvestigationArtifact") + iteration = payload.get("iteration") + if isinstance(iteration, bool) or not isinstance(iteration, int): + raise DomainError("malformed InvestigationArtifact") + try: + return cls( + artifact_id=_required_string(payload["artifact_id"], "artifact_id"), + kind=_required_string(payload["kind"], "artifact kind"), + iteration=iteration, + payload=dict(body), + created_at=_timestamp_from_text(payload["created_at"], "created_at"), + ) + except (KeyError, TypeError, ValueError) as exc: + raise DomainError("malformed InvestigationArtifact") from exc + + +@dataclass(slots=True) +class InvestigationRecord: + """Reconstructable objective, session state, and ordered decision history.""" + + objective: ResearchObjective + session: InvestigationSession + artifacts: list[InvestigationArtifact] = field(default_factory=list) + + def __post_init__(self) -> None: + if self.session.objective_id != self.objective.objective_id: + raise DomainError("session and objective identifiers do not match") + artifact_ids = [item.artifact_id for item in self.artifacts] + if len(artifact_ids) != len(set(artifact_ids)): + raise DomainError("investigation artifact identifiers must be unique") + + def append(self, kind: str, payload: Mapping[str, Any]) -> InvestigationArtifact: + artifact = InvestigationArtifact( + kind=kind, + iteration=self.session.iteration, + payload=dict(payload), + ) + self.artifacts.append(artifact) + return artifact + + def latest(self, kind: str) -> InvestigationArtifact | None: + return next((item for item in reversed(self.artifacts) if item.kind == kind), None) + + def for_iteration(self, iteration: int) -> tuple[InvestigationArtifact, ...]: + return tuple(item for item in self.artifacts if item.iteration == iteration) + + def to_dict(self) -> dict[str, object]: + return { + "objective": self.objective.to_dict(), + "session": self.session.to_dict(), + "artifacts": [item.to_dict() for item in self.artifacts], + } + + @classmethod + def from_dict(cls, payload: Mapping[str, object]) -> InvestigationRecord: + objective = payload.get("objective") + session = payload.get("session") + artifacts = payload.get("artifacts") + if not isinstance(objective, dict) or not isinstance(session, dict): + raise DomainError("malformed InvestigationRecord") + if not isinstance(artifacts, list) or any(not isinstance(item, dict) for item in artifacts): + raise DomainError("malformed InvestigationRecord artifacts") + return cls( + objective=ResearchObjective.from_dict(objective), + session=InvestigationSession.from_dict(session), + artifacts=[InvestigationArtifact.from_dict(item) for item in artifacts], + ) + + +class InvestigationRepository(Protocol): + def save(self, record: InvestigationRecord) -> None: ... + + def get(self, session_id: str) -> InvestigationRecord | None: ... + + def list(self) -> list[InvestigationRecord]: ... + + +class InMemoryInvestigationRepository: + """Deterministic adapter that round-trips JSON to expose persistence bugs.""" + + def __init__(self) -> None: + self._records: dict[str, dict[str, object]] = {} + + def save(self, record: InvestigationRecord) -> None: + self._records[record.session.session_id] = record.to_dict() + + def get(self, session_id: str) -> InvestigationRecord | None: + payload = self._records.get(session_id) + return None if payload is None else InvestigationRecord.from_dict(payload) + + def list(self) -> list[InvestigationRecord]: + return [InvestigationRecord.from_dict(self._records[key]) for key in sorted(self._records)] + + +class SQLiteInvestigationRepository: + """Durable reference adapter for session state and replayable artifacts.""" + + def __init__(self, path: str | Path) -> None: + path = Path(path) + path.parent.mkdir(parents=True, exist_ok=True) + self._connection = sqlite3.connect(str(path), check_same_thread=False) + self._connection.row_factory = sqlite3.Row + self._lock = RLock() + with self._connection: + self._connection.execute( + """ + CREATE TABLE IF NOT EXISTS investigation_sessions ( + session_id TEXT PRIMARY KEY, + objective_id TEXT NOT NULL, + state TEXT NOT NULL, + payload_json TEXT NOT NULL, + updated_at TEXT NOT NULL + ) + """ + ) + + def save(self, record: InvestigationRecord) -> None: + encoded = json.dumps(record.to_dict(), sort_keys=True, separators=(",", ":")) + with self._lock, self._connection: + self._connection.execute( + """ + INSERT INTO investigation_sessions( + session_id, objective_id, state, payload_json, updated_at + ) VALUES (?, ?, ?, ?, ?) + ON CONFLICT(session_id) DO UPDATE SET + objective_id = excluded.objective_id, + state = excluded.state, + payload_json = excluded.payload_json, + updated_at = excluded.updated_at + """, + ( + record.session.session_id, + record.objective.objective_id, + record.session.state.value, + encoded, + _timestamp_to_text(record.session.updated_at), + ), + ) + + def get(self, session_id: str) -> InvestigationRecord | None: + with self._lock: + row = self._connection.execute( + "SELECT payload_json FROM investigation_sessions WHERE session_id = ?", + (session_id,), + ).fetchone() + if row is None: + return None + payload = json.loads(str(row["payload_json"])) + if not isinstance(payload, dict): + raise DomainError("persisted investigation record is malformed") + return InvestigationRecord.from_dict(payload) + + def list(self) -> list[InvestigationRecord]: + with self._lock: + rows = self._connection.execute( + "SELECT payload_json FROM investigation_sessions ORDER BY session_id" + ).fetchall() + records: list[InvestigationRecord] = [] + for row in rows: + payload = json.loads(str(row["payload_json"])) + if not isinstance(payload, dict): + raise DomainError("persisted investigation record is malformed") + records.append(InvestigationRecord.from_dict(payload)) + return records + + def close(self) -> None: + self._connection.close() diff --git a/src/nexus_runtime/investigation/results.py b/src/nexus_runtime/investigation/results.py new file mode 100644 index 0000000..19228f3 --- /dev/null +++ b/src/nexus_runtime/investigation/results.py @@ -0,0 +1,189 @@ +"""Resolve distributed result references into structured investigation results.""" + +from __future__ import annotations + +import hashlib +import json +import sqlite3 +from pathlib import Path +from threading import RLock +from typing import Protocol + +from nexus_runtime.distributed.model import DistributedTaskState +from nexus_runtime.models import DomainError + +from .evidence import EvidenceSet, InvestigationResult, InvestigationResultState +from .execution import DistributedRuntimePort, PlanExecution + + +class InvestigationResultRepository(Protocol): + def save(self, result: InvestigationResult) -> str: ... + + def get(self, result_ref: str) -> InvestigationResult | None: ... + + +class InMemoryInvestigationResultRepository: + """Reference adapter used by harness implementations and deterministic tests.""" + + def __init__(self) -> None: + self._results: dict[str, InvestigationResult] = {} + + def save(self, result: InvestigationResult) -> str: + result_ref = f"investigation-result://{result.result_id}" + self._results[result_ref] = result + return result_ref + + def get(self, result_ref: str) -> InvestigationResult | None: + return self._results.get(result_ref) + + +class SQLiteInvestigationResultRepository: + """Durable result-reference adapter for worker/application process separation.""" + + def __init__(self, path: str | Path) -> None: + path = Path(path) + path.parent.mkdir(parents=True, exist_ok=True) + self._connection = sqlite3.connect(str(path), check_same_thread=False) + self._connection.row_factory = sqlite3.Row + self._lock = RLock() + with self._connection: + self._connection.execute( + """ + CREATE TABLE IF NOT EXISTS investigation_results ( + result_ref TEXT PRIMARY KEY, + result_id TEXT NOT NULL UNIQUE, + payload_json TEXT NOT NULL, + completed_at TEXT NOT NULL + ) + """ + ) + + def save(self, result: InvestigationResult) -> str: + result_ref = f"investigation-result://{result.result_id}" + payload = json.dumps(result.to_dict(), sort_keys=True, separators=(",", ":")) + with self._lock, self._connection: + self._connection.execute( + """ + INSERT INTO investigation_results( + result_ref, result_id, payload_json, completed_at + ) VALUES (?, ?, ?, ?) + ON CONFLICT(result_ref) DO UPDATE SET + payload_json = excluded.payload_json, + completed_at = excluded.completed_at + """, + (result_ref, result.result_id, payload, result.completed_at.isoformat()), + ) + return result_ref + + def get(self, result_ref: str) -> InvestigationResult | None: + with self._lock: + row = self._connection.execute( + "SELECT payload_json FROM investigation_results WHERE result_ref = ?", + (result_ref,), + ).fetchone() + if row is None: + return None + payload = json.loads(str(row["payload_json"])) + if not isinstance(payload, dict): + raise DomainError("persisted investigation result is malformed") + try: + return InvestigationResult.from_dict(payload) + except ValueError as exc: + raise DomainError("persisted investigation result is malformed") from exc + + def close(self) -> None: + self._connection.close() + + +class RuntimeResultCollector: + """Collect terminal task outcomes without reading coordinator internals.""" + + def __init__( + self, + runtime: DistributedRuntimePort, + results: InvestigationResultRepository, + ) -> None: + self._runtime = runtime + self._results = results + + def collect(self, execution: PlanExecution) -> tuple[InvestigationResult, ...]: + collected: list[InvestigationResult] = [] + for investigation_id, task_id in sorted(execution.task_ids.items()): + task = self._runtime.get_task(task_id) + if task.state == DistributedTaskState.SUCCEEDED: + if task.result_ref is None: + raise DomainError(f"successful task has no result reference: {task_id}") + result = self._results.get(task.result_ref) + if result is None: + raise DomainError(f"unknown investigation result reference: {task.result_ref}") + if ( + result.session_id != execution.session_id + or result.investigation_id != investigation_id + or result.task_id != task_id + or result.run_id != task.run_id + ): + raise DomainError("distributed result lineage does not match its task") + if not task.attempts or result.attempt_id != task.attempts[-1].attempt_id: + raise DomainError("distributed result attempt does not match successful task") + collected.append(result) + continue + if task.state not in { + DistributedTaskState.CANCELLED, + DistributedTaskState.DEAD_LETTERED, + }: + continue + attempt_id = task.attempts[-1].attempt_id if task.attempts else f"unattempted:{task_id}" + state = ( + InvestigationResultState.CANCELLED + if task.state == DistributedTaskState.CANCELLED + else InvestigationResultState.FAILED + ) + error = task.last_error or task.state.value.lower() + collected.append( + InvestigationResult( + session_id=execution.session_id, + investigation_id=investigation_id, + task_id=task_id, + attempt_id=attempt_id, + run_id=task.run_id, + state=state, + evidence_set=EvidenceSet( + session_id=execution.session_id, + evidence_set_id=_stable_id("evidence_set", task_id, state.value), + created_at=task.updated_at, + ), + error=error if state == InvestigationResultState.FAILED else None, + result_id=_stable_id("investigation_result", task_id, state.value), + completed_at=task.updated_at, + ) + ) + for investigation_id, reason in sorted(execution.blocked_investigations.items()): + collected.append( + InvestigationResult( + session_id=execution.session_id, + investigation_id=investigation_id, + task_id=f"blocked:{execution.plan_id}:{investigation_id}", + attempt_id="not-scheduled", + run_id=execution.run_ids[investigation_id], + state=InvestigationResultState.FAILED, + evidence_set=EvidenceSet( + session_id=execution.session_id, + evidence_set_id=_stable_id( + "evidence_set", execution.plan_id, investigation_id + ), + created_at=execution.updated_at, + ), + error=reason, + metadata={"blocked": True}, + result_id=_stable_id( + "investigation_result", execution.plan_id, investigation_id + ), + completed_at=execution.updated_at, + ) + ) + return tuple(collected) + + +def _stable_id(prefix: str, *parts: str) -> str: + digest = hashlib.sha256("\x1f".join(parts).encode("utf-8")).hexdigest()[:24] + return f"{prefix}_{digest}" diff --git a/src/nexus_runtime/investigation/session.py b/src/nexus_runtime/investigation/session.py index e4866c1..11f7e4f 100644 --- a/src/nexus_runtime/investigation/session.py +++ b/src/nexus_runtime/investigation/session.py @@ -9,17 +9,22 @@ from nexus_runtime.models import DomainError, InvalidTransition, new_id, utcnow -from .objective import _timestamp_from_text, _timestamp_to_text, _validate_timestamp +from .objective import ( + _required_string, + _timestamp_from_text, + _timestamp_to_text, + _validate_timestamp, +) def _as_int(value: object) -> int: - if isinstance(value, bool) or not isinstance(value, (int, float, str)): + if isinstance(value, bool) or not isinstance(value, int): raise TypeError - return int(value) + return value def _as_float(value: object) -> float: - if isinstance(value, bool) or not isinstance(value, (int, float, str)): + if isinstance(value, bool) or not isinstance(value, (int, float)): raise TypeError return float(value) @@ -184,7 +189,8 @@ class InvestigationSession: SessionState.CANCELLED, } ), - SessionState.PAUSED: _ACTIVE_STATES, + SessionState.PAUSED: _ACTIVE_STATES + | frozenset({SessionState.FAILED, SessionState.CANCELLED}), SessionState.COMPLETED: frozenset(), SessionState.FAILED: frozenset(), SessionState.CANCELLED: frozenset(), @@ -223,7 +229,11 @@ def transition( raise InvalidTransition( f"invalid investigation session transition: {self.state} -> {target}" ) - if self.state == SessionState.PAUSED and target != self.paused_from: + if ( + self.state == SessionState.PAUSED + and target in _ACTIVE_STATES + and target != self.paused_from + ): raise InvalidTransition("a paused session must resume its prior active state") if target in _TERMINAL_STATES and reason is None: raise DomainError("terminal transitions require a termination reason") @@ -314,19 +324,24 @@ def from_dict(cls, payload: Mapping[str, object]) -> InvestigationSession: raise TypeError reason_value = payload["termination_reason"] paused_value = payload["paused_from"] + state_value = _required_string(payload["state"], "session state") + if reason_value is not None and not isinstance(reason_value, str): + raise TypeError + if paused_value is not None and not isinstance(paused_value, str): + raise TypeError return cls( - session_id=str(payload["session_id"]), - objective_id=str(payload["objective_id"]), - state=SessionState(str(payload["state"])), + session_id=_required_string(payload["session_id"], "session_id"), + objective_id=_required_string(payload["objective_id"], "objective_id"), + state=SessionState(state_value), iteration=_as_int(payload["iteration"]), created_at=_timestamp_from_text(payload["created_at"], "created_at"), updated_at=_timestamp_from_text(payload["updated_at"], "updated_at"), budget=InvestigationBudget.from_dict(budget), usage=InvestigationUsage.from_dict(usage), termination_reason=( - None if reason_value is None else TerminationReason(str(reason_value)) + None if reason_value is None else TerminationReason(reason_value) ), - paused_from=None if paused_value is None else SessionState(str(paused_value)), + paused_from=None if paused_value is None else SessionState(paused_value), ) except (KeyError, TypeError, ValueError) as exc: raise DomainError("malformed InvestigationSession") from exc diff --git a/src/nexus_runtime/investigation/termination.py b/src/nexus_runtime/investigation/termination.py index f15d5bf..e93349b 100644 --- a/src/nexus_runtime/investigation/termination.py +++ b/src/nexus_runtime/investigation/termination.py @@ -90,7 +90,10 @@ def evaluate( SessionState.COMPLETED, "all objective success criteria are satisfied", ) - if context.objective_confidence >= self._confidence_threshold: + if ( + context.remaining_gap_count == 0 + and context.objective_confidence >= self._confidence_threshold + ): return self._stop( TerminationReason.CONFIDENCE_THRESHOLD_REACHED, SessionState.COMPLETED, diff --git a/src/nexus_runtime/investigation/verification.py b/src/nexus_runtime/investigation/verification.py index d29b20d..1348f76 100644 --- a/src/nexus_runtime/investigation/verification.py +++ b/src/nexus_runtime/investigation/verification.py @@ -5,11 +5,19 @@ from dataclasses import dataclass, field from datetime import datetime from enum import StrEnum +from typing import Any from nexus_runtime.models import new_id, utcnow from .evaluation import ClaimEvaluation, Evaluation -from .evidence import ClaimStatement +from .evidence import ( + ClaimStatement, + _parse_timestamp, + _persisted_bool, + _persisted_float, + _persisted_string, + _persisted_strings, +) class EpistemicStatus(StrEnum): @@ -53,6 +61,50 @@ class VerificationDecision: unresolved_conflict_ids: tuple[str, ...] reasons: tuple[str, ...] + def to_dict(self) -> dict[str, Any]: + return { + "claim": self.claim.to_dict(), + "status": self.status.value, + "eligible_for_update": self.eligible_for_update, + "confidence": self.confidence, + "supporting_evidence_ids": list(self.supporting_evidence_ids), + "contradicting_evidence_ids": list(self.contradicting_evidence_ids), + "unresolved_conflict_ids": list(self.unresolved_conflict_ids), + "reasons": list(self.reasons), + } + + @classmethod + def from_dict(cls, payload: dict[str, Any]) -> VerificationDecision: + claim = payload.get("claim") + sequence_fields = ( + "supporting_evidence_ids", + "contradicting_evidence_ids", + "unresolved_conflict_ids", + "reasons", + ) + if not isinstance(claim, dict) or any( + not isinstance(payload.get(name), list) for name in sequence_fields + ): + raise ValueError("malformed verification decision") + status = payload.get("status") + if not isinstance(status, str): + raise ValueError("malformed verification decision") + try: + return cls( + claim=ClaimStatement.from_dict(claim), + status=EpistemicStatus(status), + eligible_for_update=_persisted_bool(payload, "eligible_for_update"), + confidence=_persisted_float(payload, "confidence"), + supporting_evidence_ids=_persisted_strings(payload, "supporting_evidence_ids"), + contradicting_evidence_ids=_persisted_strings( + payload, "contradicting_evidence_ids" + ), + unresolved_conflict_ids=_persisted_strings(payload, "unresolved_conflict_ids"), + reasons=_persisted_strings(payload, "reasons"), + ) + except (KeyError, TypeError, ValueError) as exc: + raise ValueError("malformed verification decision") from exc + @dataclass(frozen=True, slots=True) class VerificationReport: @@ -66,6 +118,31 @@ class VerificationReport: def eligible_claims(self) -> tuple[VerificationDecision, ...]: return tuple(decision for decision in self.decisions if decision.eligible_for_update) + def to_dict(self) -> dict[str, Any]: + return { + "verification_id": self.verification_id, + "session_id": self.session_id, + "evaluation_id": self.evaluation_id, + "decisions": [item.to_dict() for item in self.decisions], + "verified_at": self.verified_at.isoformat(), + } + + @classmethod + def from_dict(cls, payload: dict[str, Any]) -> VerificationReport: + decisions = payload.get("decisions") + if not isinstance(decisions, list) or any(not isinstance(item, dict) for item in decisions): + raise ValueError("malformed verification report") + try: + return cls( + verification_id=_persisted_string(payload, "verification_id"), + session_id=_persisted_string(payload, "session_id"), + evaluation_id=_persisted_string(payload, "evaluation_id"), + decisions=tuple(VerificationDecision.from_dict(item) for item in decisions), + verified_at=_parse_timestamp(payload["verified_at"]), + ) + except (KeyError, TypeError, ValueError) as exc: + raise ValueError("malformed verification report") from exc + class ClaimVerifier: """Require provenance, independent sources, quality, and no conflict.""" diff --git a/tests/test_evidence_loop.py b/tests/test_evidence_loop.py index 179cabc..25e4c0f 100644 --- a/tests/test_evidence_loop.py +++ b/tests/test_evidence_loop.py @@ -28,6 +28,7 @@ from nexus_runtime.investigation.verification import ( ClaimVerifier, EpistemicStatus, + VerificationDecision, VerificationPolicy, ) @@ -112,6 +113,42 @@ def test_incomplete_provenance_is_rejected(self): with pytest.raises(ValueError, match="tool_call_id"): replace(_provenance(), tool_call_id="") + def test_malformed_persisted_provenance_is_not_coerced(self): + payload = _provenance().to_dict() + payload["attempt_id"] = None + with pytest.raises(ValueError, match="attempt_id"): + EvidenceProvenance.from_dict(payload) + + def test_malformed_persisted_verification_boolean_is_not_coerced(self): + evaluation = EvidenceEvaluator().evaluate( + EvidenceSet(session_id="session-1", evidence=(_evidence(),)) + ) + payload = ( + ClaimVerifier(VerificationPolicy(min_independent_sources=1)) + .verify(evaluation) + .decisions[0] + .to_dict() + ) + payload["eligible_for_update"] = "false" + with pytest.raises(ValueError, match="malformed verification decision"): + VerificationDecision.from_dict(payload) + + def test_claim_id_cannot_alias_different_claims(self): + london = ClaimStatement("Atlas in London", "Atlas", "located_in", "London", "claim-x") + paris = ClaimStatement("Atlas in Paris", "Atlas", "located_in", "Paris", "claim-x") + with pytest.raises(ValueError, match="claim_id"): + EvidenceSet( + "session-1", + ( + _evidence(claim=london), + _evidence( + claim=paris, + source_id="source-b", + source_reference="https://example.test/b", + ), + ), + ) + def test_anonymous_evidence_is_rejected(self): with pytest.raises(ValueError, match="excerpt or payload"): replace(_evidence(), excerpt="", payload={}) @@ -343,6 +380,26 @@ def test_invalid_update_is_not_submitted(self, ingested_engine): assert result.accepted_records == 0 assert ingested_engine.repository.claims.count() == before + def test_dangling_source_lineage_is_rejected(self, ingested_engine): + claim = ClaimStatement("Atlas in London", "Atlas", "located_in", "London") + evidence_set = EvidenceSet( + "session-1", + ( + _evidence(claim=claim), + _evidence( + claim=claim, + source_id="source-b", + source_reference="https://example.test/b", + document_id="document-b", + chunk_id="chunk-b", + ), + ), + ) + report = ClaimVerifier().verify(EvidenceEvaluator().evaluate(evidence_set)) + + with pytest.raises(ValueError, match="does not resolve"): + KnowledgeUpdateIntegrator(ingested_engine).prepare(report, evidence_set) + def test_existing_contradiction_is_preserved_and_not_auto_verified(self, ingested_engine): claim, evidence_set = self._verified_set(ingested_engine) existing = Claim( diff --git a/tests/test_investigation_application.py b/tests/test_investigation_application.py new file mode 100644 index 0000000..46e25bd --- /dev/null +++ b/tests/test_investigation_application.py @@ -0,0 +1,696 @@ +from __future__ import annotations + +from dataclasses import replace +from datetime import UTC, datetime, timedelta +from pathlib import Path +from types import SimpleNamespace + +import pytest + +from nexus_knowledge.domain.common import VerificationState +from nexus_knowledge.domain.knowledge_gap import Investigation, KnowledgeGap +from nexus_knowledge.knowledge.uncertainty import UncertaintyAssessment +from nexus_knowledge.service.engine import KnowledgeUpdate, KnowledgeUpdateReceipt +from nexus_runtime.distributed.model import FailureClass, RetryPolicy +from nexus_runtime.distributed.service import RuntimeApplication +from nexus_runtime.distributed.simulator import LocalDistributedSimulator +from nexus_runtime.distributed.worker import ( + HarnessExecutionContext, + HarnessOutcome, + HarnessStatus, +) +from nexus_runtime.events import InMemoryEventBus +from nexus_runtime.investigation.application import InvestigationApplication +from nexus_runtime.investigation.evidence import ( + ClaimStatement, + Evidence, + EvidenceSet, + InvestigationResult, + InvestigationResultState, +) +from nexus_runtime.investigation.execution import PlanExecution, PlanExecutionController +from nexus_runtime.investigation.metrics import InMemoryInvestigationMetrics +from nexus_runtime.investigation.objective import ResearchObjective +from nexus_runtime.investigation.provenance import EvidenceProvenance +from nexus_runtime.investigation.repository import ( + InMemoryInvestigationRepository, + SQLiteInvestigationRepository, +) +from nexus_runtime.investigation.results import ( + InMemoryInvestigationResultRepository, + RuntimeResultCollector, + SQLiteInvestigationResultRepository, +) +from nexus_runtime.investigation.session import ( + InvestigationBudget, + SessionState, + TerminationReason, +) +from nexus_runtime.models import DomainError + +NOW = datetime(2026, 1, 1, tzinfo=UTC) + + +def budget(max_iterations: int = 3) -> InvestigationBudget: + return InvestigationBudget(max_iterations, 20, 20, 100.0, timedelta(hours=1)) + + +def objective() -> ResearchObjective: + return ResearchObjective( + question="What is Acme's verified status?", + success_criteria=("status has two independent sources",), + objective_id="objective-app", + created_at=NOW, + ) + + +def gap(index: int = 0) -> KnowledgeGap: + item = KnowledgeGap( + id=f"gap-{index}", + kind="missing_evidence", + description=f"verify Acme status {index}", + reason="no independent evidence", + affected_entities=["acme"], + uncertainty=0.9, + importance=0.9, + estimated_cost=1.0, + created_at="2026-01-01T00:00:00Z", + ) + item.candidate_investigations = [ + Investigation( + id=f"legacy-{index}", + gap_id=item.id, + description=f"find independent evidence for Acme {index}", + target_entities=["acme"], + estimated_cost=1.0, + created_at="2026-01-01T00:00:00Z", + ) + ] + return item + + +class FakeKnowledge: + def __init__(self, gaps: list[KnowledgeGap] | None = None) -> None: + self.gaps = [gap()] if gaps is None else gaps + self.updated = False + self.committed: list[KnowledgeUpdate] = [] + self.fail_commit_once = False + + def retrieve(self, query: str, top_k: int = 10) -> object: + candidate = SimpleNamespace(chunk_id="chunk-existing") + return SimpleNamespace(request_id="request-1", candidates=[candidate]) + + def graphrag(self, query: str, top_k: int = 8, depth: int = 2) -> object: + entity = SimpleNamespace(id="acme") + return SimpleNamespace(entities=[entity], relations=[], confidence=0.4) + + def find_knowledge_gaps(self) -> list[KnowledgeGap]: + return [] if self.updated else self.gaps + + def detect_contradictions(self) -> list[object]: + return [] + + def commit_knowledge_update(self, update: KnowledgeUpdate) -> KnowledgeUpdateReceipt: + if self.fail_commit_once: + self.fail_commit_once = False + raise RuntimeError("injected knowledge commit failure") + self.committed.append(update) + self.updated = bool(update.claims) + return KnowledgeUpdateReceipt( + accepted=len(update.claims) + len(update.evidence), rejected=0, errors=[] + ) + + def verify_claim(self, claim_id: str) -> UncertaintyAssessment: + return UncertaintyAssessment( + claim_id=claim_id, + confidence=0.9, + uncertainty=0.1, + verification_state=VerificationState.VERIFIED, + supporting_evidence_count=2, + contradicting_evidence_count=0, + source_quality=0.9, + source_diversity=1.0, + recency=1.0, + components={}, + ) + + def validate_evidence_provenance( + self, + source_id: str, + document_id: str, + chunk_id: str, + source_reference: str, + ) -> bool: + return all((source_id, document_id, chunk_id, source_reference)) + + +class EvidenceHarness: + def __init__( + self, + results: InMemoryInvestigationResultRepository, + fail_once: bool = False, + ) -> None: + self.result_repository = results + self.fail_once = fail_once + self.calls = 0 + self.results: list[InvestigationResult] = [] + + def execute_or_resume( + self, + context: HarnessExecutionContext, + cancellation_requested, + ): + self.calls += 1 + if self.fail_once and self.calls == 1: + return HarnessOutcome( + HarnessStatus.FAILED, + checkpoint_ref="checkpoint://evidence-agent", + failure_class=FailureClass.TRANSIENT, + error="temporary source failure", + ) + task = context + investigation_id = str(context.metadata["investigation_id"]) + claim = ClaimStatement( + text="Acme is active", + subject="Acme", + predicate="status", + object="active", + claim_id="claim-acme-active", + ) + evidence = tuple( + Evidence( + investigation_id=investigation_id, + source=f"source://{source}", + claim=claim, + provenance=EvidenceProvenance( + session_id=context.correlation_id, + investigation_id=investigation_id, + task_id=task.task_id, + attempt_id=context.attempt_id, + run_id=context.run_id, + tool_call_id=f"tool-{source}", + source_id=f"source-{source}", + document_id=f"document-{source}", + chunk_id=f"chunk-{source}", + source_reference=f"source://{source}", + ), + confidence=0.95, + source_quality=0.9, + excerpt=f"independent report {source}", + evidence_id=f"evidence-{context.task_id}-{source}", + ) + for source in ("a", "b") + ) + evidence_set = EvidenceSet(session_id=context.correlation_id, evidence=evidence) + result = InvestigationResult( + session_id=context.correlation_id, + investigation_id=investigation_id, + task_id=context.task_id, + attempt_id=context.attempt_id, + run_id=context.run_id, + state=InvestigationResultState.COMPLETED, + evidence_set=evidence_set, + ) + self.results.append(result) + return HarnessOutcome(HarnessStatus.SUCCEEDED, self.result_repository.save(result)) + + def cancel_run(self, run_id: str) -> None: + return None + + +def make_runtime(fail_once: bool = False): + simulator = LocalDistributedSimulator(start=NOW) + results = InMemoryInvestigationResultRepository() + harness = EvidenceHarness(results, fail_once=fail_once) + simulator.add_worker( + "evidence-worker", + frozenset({"document_analysis", "search"}), + harness, + ) + runtime = RuntimeApplication(simulator.coordinator) + return ( + simulator, + harness, + PlanExecutionController(runtime), + RuntimeResultCollector(runtime, results), + ) + + +def test_closed_loop_resolves_gap_and_completes_objective() -> None: + knowledge = FakeKnowledge() + simulator, _, execution, collector = make_runtime() + events = InMemoryEventBus() + metrics = InMemoryInvestigationMetrics() + app = InvestigationApplication( + knowledge, + repository=InMemoryInvestigationRepository(), + execution=execution, + result_collector=collector, + event_bus=events, + metrics=metrics, + ) + session = app.create(objective(), budget()) + planning = app.plan_iteration(session.session_id, worker_capacity=1) + assert planning.plan is not None + investigation_id = planning.plan.investigations[0].investigation_id + app.start_execution( + session.session_id, + planning.plan, + {investigation_id: "agent-run-acme"}, + ) + + report = simulator.run_until_terminal() + results = app.collect_execution_results(session.session_id) + outcome = app.process_results( + session.session_id, + results, + objective_satisfied=True, + ) + + record = app.status(session.session_id) + assert report.succeeded == 1 + assert outcome.verification.eligible_claims + assert outcome.update_result.committed_claim_ids == ("claim-acme-active",) + assert outcome.progress.resolved_gap_ids == ("gap-0",) + assert record.session.state == SessionState.COMPLETED + assert record.session.iteration == 1 + assert record.session.termination_reason == TerminationReason.OBJECTIVE_SATISFIED + event_types = {item.event_type for item in events.events} + assert { + "investigation.session_created", + "investigation.plan_created", + "investigation.execution_started", + "investigation.evidence_collected", + "investigation.knowledge_updated", + "investigation.completed", + } <= event_types + snapshot = metrics.snapshot() + assert snapshot["counters"]["gaps_resolved"] == 1 + assert snapshot["gauges"]["investigation_success_rate"] == 1.0 + + +def test_distributed_retry_preserves_attempt_lineage() -> None: + knowledge = FakeKnowledge() + simulator, harness, execution, _ = make_runtime(fail_once=True) + app = InvestigationApplication(knowledge, execution=execution) + session = app.create(objective(), budget()) + planning = app.plan_iteration(session.session_id, worker_capacity=1) + assert planning.plan is not None + investigation_id = planning.plan.investigations[0].investigation_id + status = app.start_execution( + session.session_id, + planning.plan, + {investigation_id: "agent-run-retry"}, + ) + task_id = status.execution.task_ids[investigation_id] + task = simulator.coordinator.require_task(task_id) + task.retry_policy = RetryPolicy( + max_attempts=2, + initial_backoff=timedelta(0), + max_backoff=timedelta(0), + ) + + report = simulator.run_until_terminal() + + completed = simulator.coordinator.require_task(task_id) + assert report.retries == 1 + assert completed.attempt == 2 + assert harness.results[0].attempt_id == completed.attempts[-1].attempt_id + evidence_attempt = harness.results[0].evidence_set.evidence[0].provenance.attempt_id + assert evidence_attempt == completed.attempts[-1].attempt_id + + +def test_worker_loss_reassigns_investigation_without_corrupting_session() -> None: + knowledge = FakeKnowledge() + simulator, harness, execution, collector = make_runtime() + app = InvestigationApplication( + knowledge, + execution=execution, + result_collector=collector, + ) + session = app.create(objective(), budget()) + planning = app.plan_iteration(session.session_id, worker_capacity=1) + assert planning.plan is not None + investigation_id = planning.plan.investigations[0].investigation_id + started = app.start_execution( + session.session_id, + planning.plan, + {investigation_id: "agent-run-worker-loss"}, + ) + task_id = started.execution.task_ids[investigation_id] + crashed = simulator.workers[0] + claimed = simulator.coordinator.claim_task(crashed.identity) + assert claimed is not None and claimed.lease_id is not None + simulator.coordinator.start_task(crashed.identity, task_id, claimed.lease_id) + crashed.crash() + + simulator.clock.advance(timedelta(seconds=31)) + simulator.coordinator.recover() + simulator.clock.advance(timedelta(seconds=1)) + simulator.coordinator.recover() + replacement = simulator.add_worker( + "replacement-worker", + frozenset({"document_analysis", "search"}), + harness, + ) + completed = replacement.poll_once() + assert completed is not None + results = app.collect_execution_results(session.session_id) + outcome = app.process_results( + session.session_id, + results, + objective_satisfied=True, + ) + + assert completed.attempt == 2 + assert completed.attempts[0].error == "lease expired" + assert results[0].attempt_id == completed.attempts[-1].attempt_id + assert outcome.update_result.committed_claim_ids == ("claim-acme-active",) + assert app.status(session.session_id).session.state == SessionState.COMPLETED + + +def test_plan_dependencies_submit_in_ready_waves() -> None: + knowledge = FakeKnowledge([gap(0), gap(1)]) + simulator, _, execution, _ = make_runtime() + app = InvestigationApplication(knowledge, execution=execution) + session = app.create(objective(), budget()) + snapshot = app.observe(session.session_id) + candidates = app.generate(session.session_id, snapshot) + scores = app.score(session.session_id, snapshot, candidates) + selection = app.select(session.session_id, scores, worker_capacity=2) + ids = [item.candidate.investigation_id for item in selection.selected] + plan = app.build_plan(session.session_id, selection, dependencies={ids[1]: (ids[0],)}) + + started = app.start_execution( + session.session_id, + plan, + {ids[0]: "run-root", ids[1]: "run-child"}, + ) + assert set(started.execution.task_ids) == {ids[0]} + simulator.workers[0].poll_once() + advanced = app.advance_execution(session.session_id) + assert set(advanced.execution.task_ids) == set(ids) + + +def test_ready_wave_recovery_reuses_task_after_lost_local_ack() -> None: + knowledge = FakeKnowledge() + simulator = LocalDistributedSimulator(start=NOW) + runtime = RuntimeApplication(simulator.coordinator) + controller = PlanExecutionController(runtime) + app = InvestigationApplication(knowledge) + session = app.create(objective(), budget()) + planning = app.plan_iteration(session.session_id, worker_capacity=1) + assert planning.plan is not None + investigation_id = planning.plan.investigations[0].investigation_id + run_ids = {investigation_id: "run-idempotent-submit"} + lost_ack_execution = controller.prepare(planning.plan, run_ids) + first = controller.advance(planning.plan, lost_ack_execution) + recovered_intent = controller.prepare(planning.plan, run_ids) + + recovered = controller.advance(planning.plan, recovered_intent) + + assert recovered.execution.task_ids == first.execution.task_ids + assert len(runtime.list_tasks()) == 1 + + +def test_blocked_investigation_becomes_explicit_failed_result() -> None: + simulator = LocalDistributedSimulator(start=NOW) + runtime = RuntimeApplication(simulator.coordinator) + collector = RuntimeResultCollector(runtime, InMemoryInvestigationResultRepository()) + execution = PlanExecution( + plan_id="plan-blocked", + session_id="session-blocked", + run_ids={"investigation-blocked": "run-blocked"}, + blocked_investigations={"investigation-blocked": "dependency failed: parent"}, + ) + + results = collector.collect(execution) + + assert results[0].state == InvestigationResultState.FAILED + assert results[0].metadata == {"blocked": True} + assert results[0].error == "dependency failed: parent" + + +def test_cancelled_task_yields_partial_iteration_then_replans() -> None: + knowledge = FakeKnowledge() + simulator = LocalDistributedSimulator(start=NOW) + runtime = RuntimeApplication(simulator.coordinator) + result_repository = InMemoryInvestigationResultRepository() + app = InvestigationApplication( + knowledge, + execution=PlanExecutionController(runtime), + result_collector=RuntimeResultCollector(runtime, result_repository), + ) + session = app.create(objective(), budget()) + planning = app.plan_iteration(session.session_id, worker_capacity=1) + assert planning.plan is not None + investigation_id = planning.plan.investigations[0].investigation_id + started = app.start_execution( + session.session_id, + planning.plan, + {investigation_id: "run-cancelled-task"}, + ) + task_id = started.execution.task_ids[investigation_id] + runtime.cancel_task(task_id, "failure-injection") + + results = app.collect_execution_results(session.session_id) + first_iteration = app.process_results(session.session_id, results) + second_planning = app.plan_iteration(session.session_id, worker_capacity=1) + + assert results[0].state == InvestigationResultState.CANCELLED + assert not first_iteration.termination.terminate + assert second_planning.plan is not None + assert app.status(session.session_id).session.iteration == 1 + with pytest.raises(DomainError, match="current iteration plan"): + app.start_execution( + session.session_id, + planning.plan, + {investigation_id: "stale-run"}, + ) + + +def test_sqlite_pause_restart_resume_preserves_artifacts(tmp_path: Path) -> None: + path = tmp_path / "investigations.sqlite" + first_repository = SQLiteInvestigationRepository(path) + first = InvestigationApplication(FakeKnowledge(), repository=first_repository) + session = first.create(objective(), budget()) + first.observe(session.session_id) + first.pause(session.session_id) + artifact_count = len(first.status(session.session_id).artifacts) + first_repository.close() + + resumed_repository = SQLiteInvestigationRepository(path) + resumed = InvestigationApplication(FakeKnowledge(), repository=resumed_repository) + state = resumed.resume(session.session_id) + + assert state.state == SessionState.PLANNING + assert len(resumed.status(session.session_id).artifacts) == artifact_count + 1 + resumed_repository.close() + + +def test_restart_after_collection_resumes_without_rerunning_agents(tmp_path: Path) -> None: + path = tmp_path / "collected.sqlite" + knowledge = FakeKnowledge() + simulator, harness, execution, collector = make_runtime() + first_repository = SQLiteInvestigationRepository(path) + first = InvestigationApplication( + knowledge, + repository=first_repository, + execution=execution, + result_collector=collector, + ) + session = first.create(objective(), budget()) + planning = first.plan_iteration(session.session_id, worker_capacity=1) + assert planning.plan is not None + investigation_id = planning.plan.investigations[0].investigation_id + first.start_execution( + session.session_id, + planning.plan, + {investigation_id: "agent-run-resume"}, + ) + simulator.run_until_terminal() + first.collect_evidence(session.session_id, first.collect_execution_results(session.session_id)) + calls_before_restart = harness.calls + first_repository.close() + + resumed_repository = SQLiteInvestigationRepository(path) + resumed = InvestigationApplication(knowledge, repository=resumed_repository) + stored = resumed.stored_results(session.session_id) + outcome = resumed.resume_collected_iteration( + session.session_id, + objective_satisfied=True, + ) + + assert stored[0].result_id == harness.results[0].result_id + assert outcome.update_result.committed_claim_ids == ("claim-acme-active",) + assert resumed.status(session.session_id).session.iteration == 1 + assert harness.calls == calls_before_restart + resumed_repository.close() + + +def test_result_reference_survives_repository_restart(tmp_path: Path) -> None: + simulator, harness, execution, _ = make_runtime() + knowledge = FakeKnowledge() + app = InvestigationApplication(knowledge, execution=execution) + session = app.create(objective(), budget()) + planning = app.plan_iteration(session.session_id, worker_capacity=1) + assert planning.plan is not None + investigation_id = planning.plan.investigations[0].investigation_id + app.start_execution( + session.session_id, + planning.plan, + {investigation_id: "agent-run-durable-result"}, + ) + simulator.run_until_terminal() + result = harness.results[0] + path = tmp_path / "results.sqlite" + first = SQLiteInvestigationResultRepository(path) + result_ref = first.save(result) + first.close() + + resumed = SQLiteInvestigationResultRepository(path) + restored = resumed.get(result_ref) + + assert restored is not None + assert restored.to_dict() == result.to_dict() + resumed.close() + + +def test_restart_from_evaluating_reuses_verified_evidence(tmp_path: Path) -> None: + path = tmp_path / "evaluating.sqlite" + knowledge = FakeKnowledge() + simulator, _, execution, collector = make_runtime() + first_repository = SQLiteInvestigationRepository(path) + first = InvestigationApplication( + knowledge, + repository=first_repository, + execution=execution, + result_collector=collector, + ) + session = first.create(objective(), budget()) + planning = first.plan_iteration(session.session_id, worker_capacity=1) + assert planning.plan is not None + investigation_id = planning.plan.investigations[0].investigation_id + first.start_execution( + session.session_id, + planning.plan, + {investigation_id: "agent-run-evaluating-resume"}, + ) + simulator.run_until_terminal() + evidence_set = first.collect_evidence( + session.session_id, + first.collect_execution_results(session.session_id), + ) + evaluation = first.evaluate(session.session_id, evidence_set) + first.verify(session.session_id, evaluation) + assert first.status(session.session_id).session.state == SessionState.EVALUATING + first_repository.close() + + resumed_repository = SQLiteInvestigationRepository(path) + resumed = InvestigationApplication(knowledge, repository=resumed_repository) + outcome = resumed.resume_iteration(session.session_id, objective_satisfied=True) + + assert outcome.update_result.committed_claim_ids == ("claim-acme-active",) + assert resumed.status(session.session_id).session.state == SessionState.COMPLETED + resumed_repository.close() + + +def test_restart_from_pending_update_replays_idempotent_claim_ids(tmp_path: Path) -> None: + path = tmp_path / "updating.sqlite" + knowledge = FakeKnowledge() + simulator, _, execution, collector = make_runtime() + first_repository = SQLiteInvestigationRepository(path) + first = InvestigationApplication( + knowledge, + repository=first_repository, + execution=execution, + result_collector=collector, + ) + session = first.create(objective(), budget()) + planning = first.plan_iteration(session.session_id, worker_capacity=1) + assert planning.plan is not None + investigation_id = planning.plan.investigations[0].investigation_id + first.start_execution( + session.session_id, + planning.plan, + {investigation_id: "agent-run-update-resume"}, + ) + simulator.run_until_terminal() + evidence_set = first.collect_evidence( + session.session_id, + first.collect_execution_results(session.session_id), + ) + evaluation = first.evaluate(session.session_id, evidence_set) + verification = first.verify(session.session_id, evaluation) + knowledge.fail_commit_once = True + with pytest.raises(RuntimeError, match="injected"): + first.update_knowledge(session.session_id, verification, evidence_set) + assert first.status(session.session_id).session.state == SessionState.UPDATING + assert first.status(session.session_id).latest("knowledge_update") is not None + first_repository.close() + + resumed_repository = SQLiteInvestigationRepository(path) + resumed = InvestigationApplication(knowledge, repository=resumed_repository) + outcome = resumed.resume_iteration(session.session_id, objective_satisfied=True) + + assert outcome.update_result.committed_claim_ids == ("claim-acme-active",) + assert resumed.status(session.session_id).session.state == SessionState.COMPLETED + resumed_repository.close() + + +def test_no_gaps_terminates_without_spawning_agents() -> None: + app = InvestigationApplication(FakeKnowledge([])) + session = app.create(objective(), budget()) + planning = app.plan_iteration(session.session_id, worker_capacity=2) + + assert planning.plan is None + assert planning.termination is not None + assert app.status(session.session_id).session.state == SessionState.COMPLETED + + +def test_planning_rejects_modified_copy_of_persisted_snapshot() -> None: + app = InvestigationApplication(FakeKnowledge()) + session = app.create(objective(), budget()) + snapshot = app.observe(session.session_id) + + with pytest.raises(DomainError, match="not active"): + app.generate(session.session_id, replace(snapshot, summary="forged summary")) + + +def test_paused_session_can_be_cancelled() -> None: + app = InvestigationApplication(FakeKnowledge()) + session = app.create(objective(), budget()) + app.pause(session.session_id) + + cancelled = app.cancel(session.session_id) + + assert cancelled.state == SessionState.CANCELLED + assert cancelled.termination_reason == TerminationReason.USER_CANCELLATION + + +def test_cancelled_session_cannot_submit_more_task_waves() -> None: + knowledge = FakeKnowledge([gap(0), gap(1)]) + simulator = LocalDistributedSimulator(start=NOW) + runtime = RuntimeApplication(simulator.coordinator) + app = InvestigationApplication( + knowledge, + execution=PlanExecutionController(runtime), + ) + session = app.create(objective(), budget()) + snapshot = app.observe(session.session_id) + candidates = app.generate(session.session_id, snapshot) + scores = app.score(session.session_id, snapshot, candidates) + selection = app.select(session.session_id, scores, worker_capacity=2) + ids = [item.candidate.investigation_id for item in selection.selected] + plan = app.build_plan(session.session_id, selection, dependencies={ids[1]: (ids[0],)}) + app.start_execution( + session.session_id, + plan, + {ids[0]: "run-cancel-root", ids[1]: "run-cancel-child"}, + ) + app.cancel(session.session_id) + + with pytest.raises(DomainError, match="only while EXECUTING"): + app.advance_execution(session.session_id) + + assert len(runtime.list_tasks()) == 1 diff --git a/tests/test_investigation_cli.py b/tests/test_investigation_cli.py new file mode 100644 index 0000000..5aaefb7 --- /dev/null +++ b/tests/test_investigation_cli.py @@ -0,0 +1,56 @@ +from __future__ import annotations + +import json + +from nexus_runtime.investigation.application import InvestigationApplication +from nexus_runtime.investigation.benchmark import run_benchmark +from nexus_runtime.investigation.cli import main +from nexus_runtime.investigation.repository import InMemoryInvestigationRepository + +from .test_investigation_application import FakeKnowledge + + +def test_research_cli_create_status_pause_resume_and_cancel(capsys) -> None: + app = InvestigationApplication(FakeKnowledge(), repository=InMemoryInvestigationRepository()) + assert main(["create", "What is Acme?", "--criterion", "verified"], application=app) == 0 + created = json.loads(capsys.readouterr().out) + session_id = created["session_id"] + + assert main(["status", session_id], application=app) == 0 + assert json.loads(capsys.readouterr().out)["session"]["state"] == "PLANNING" + assert main(["pause", session_id], application=app) == 0 + assert json.loads(capsys.readouterr().out)["state"] == "PAUSED" + assert main(["resume", session_id], application=app) == 0 + assert json.loads(capsys.readouterr().out)["state"] == "PLANNING" + assert main(["cancel", session_id], application=app) == 0 + assert json.loads(capsys.readouterr().out)["state"] == "CANCELLED" + + +def test_research_cli_explain_gaps_plan_and_iterations(capsys) -> None: + app = InvestigationApplication(FakeKnowledge()) + assert main(["create", "What is Acme?", "--criterion", "verified"], application=app) == 0 + session_id = json.loads(capsys.readouterr().out)["session_id"] + app.plan_iteration(session_id, worker_capacity=1) + + assert main(["gaps", session_id], application=app) == 0 + assert json.loads(capsys.readouterr().out)[0]["id"] == "gap-0" + assert main(["plan", session_id], application=app) == 0 + assert json.loads(capsys.readouterr().out)["session_id"] == session_id + assert main(["iterations", session_id], application=app) == 0 + assert json.loads(capsys.readouterr().out)[0]["iteration"] == 0 + assert main(["explain", session_id], application=app) == 0 + explanation = json.loads(capsys.readouterr().out) + assert explanation["why_investigated"] + + +def test_investigation_benchmark_has_required_workload() -> None: + report = run_benchmark() + assert report["methodology"] == { + "gaps": 10, + "candidates": 50, + "selected": 10, + "distributed_tasks": 10, + "clock": "fixed UTC inputs with local deterministic runtime", + } + assert report["results"]["succeeded_tasks"] == 10 + assert report["results"]["accepted_evidence"] == 10 diff --git a/tests/test_investigation_intelligence.py b/tests/test_investigation_intelligence.py index f289337..5fbf338 100644 --- a/tests/test_investigation_intelligence.py +++ b/tests/test_investigation_intelligence.py @@ -443,7 +443,7 @@ def session(self, **budget_overrides: object) -> InvestigationSession: TerminationReason.OBJECTIVE_SATISFIED, ), ( - TerminationContext(False, 0.95, 1, 0.5), + TerminationContext(False, 0.95, 0, 0.5), TerminationReason.CONFIDENCE_THRESHOLD_REACHED, ), ( @@ -496,3 +496,9 @@ def test_valuable_work_continues(self) -> None: ) assert not decision.terminate assert decision.reason is None + + def test_confidence_does_not_hide_remaining_gaps(self) -> None: + decision = TerminationPolicy().evaluate( + self.session(), TerminationContext(False, 0.99, 1, 0.5) + ) + assert not decision.terminate