build(agent): new-agents-3#dd492b iteration
This commit is contained in:
parent
5f86c216a6
commit
a443cdc797
|
|
@ -0,0 +1,21 @@
|
|||
node_modules/
|
||||
.npmrc
|
||||
.env
|
||||
.env.*
|
||||
__tests__/
|
||||
coverage/
|
||||
.nyc_output/
|
||||
dist/
|
||||
build/
|
||||
.cache/
|
||||
*.log
|
||||
.DS_Store
|
||||
tmp/
|
||||
.tmp/
|
||||
__pycache__/
|
||||
*.pyc
|
||||
.venv/
|
||||
venv/
|
||||
*.egg-info/
|
||||
.pytest_cache/
|
||||
READY_TO_PUBLISH
|
||||
|
|
@ -0,0 +1,29 @@
|
|||
# MissionLedger Agents Guide
|
||||
|
||||
Overview
|
||||
- MissionLedger is a production-grade core for verifiable, contract-driven planning in intermittently-connected fleets.
|
||||
- This repo contains a minimal but production-oriented MVP to bootstrap the ecosystem: a DSL for per-asset LocalProblems, safety contracts, a shadow planner, a lightweight safety verifier, and a governanced ledger.
|
||||
|
||||
Architecture (high level)
|
||||
- missionledger.dsl: Defines the LocalProblem, SafetyContracts, PrivacyContracts, and PlanDelta data models.
|
||||
- missionledger.planner: ShadowPlanner that proposes a safe alternative plan in parallel to the executor.
|
||||
- missionledger.safety: Lightweight safety verifier used by both the main planner and shadow planner.
|
||||
- missionledger.governance: Tamper-evident governance ledger with hash chaining for auditability.
|
||||
- missionledger.adapter (placeholder): SDK adapters for various assets and runtime stacks (rover, drone, habitat, orbiters).
|
||||
|
||||
Tech stack (current MVP)
|
||||
- Python 3.8+ with a small, typed core using dataclasses.
|
||||
- No external runtime dependencies for MVP; easy to extend with real solvers later.
|
||||
|
||||
Development commands
|
||||
- Run tests: bash test.sh
|
||||
- Build the package: python3 -m build
|
||||
- Lint/type checks: not included in MVP yet; plan to add flake8/ruff and mypy in future iterations.
|
||||
|
||||
Tests and verification
|
||||
- The test suite exercises the LocalProblem DSL, the ShadowPlanner, the safety verifier, and the governance ledger entries.
|
||||
- Extend tests to cover full end-to-end delta-sync, proofs, and HIL adapters in later phases.
|
||||
|
||||
Contribution guidelines
|
||||
- Follow the architecture layout described above.
|
||||
- Add tests for any new feature and ensure tests pass locally before proposing PRs.
|
||||
22
README.md
22
README.md
|
|
@ -1,3 +1,21 @@
|
|||
# idea19-missionledger-verifiable-contract
|
||||
# MissionLedger: Verifiable, Contract-Driven AI Mission Planner (MVP)
|
||||
|
||||
Source logic for Idea #19
|
||||
MissionLedger provides a minimal, production-ready core for contract-driven planning on intermittently-connected fleets. This MVP focuses on a clean, testable Python implementation of the DSL, a shadow planner, a lightweight safety verifier, and a tamper-evident governance ledger.
|
||||
|
||||
What’s included in this MVP
|
||||
- LocalProblem DSL: per-asset objectives, variables, constraints, and risk budgets.
|
||||
- SafetyContracts and PrivacyContracts scaffolding for future integration with verifiable proofs and privacy controls.
|
||||
- ShadowPlanner: a simple, parallel planning agent that generates a safe delta alongside the primary plan.
|
||||
- Lightweight safety verifier: deterministic checks for plan validity in the MVP.
|
||||
- GovernanceLedger: append-only, tamper-evident log with hash chaining for auditability.
|
||||
- A minimal tests suite to verify the core components.
|
||||
|
||||
Getting started
|
||||
- Install dependencies and run tests via the provided test script.
|
||||
- Build the package with the standard Python tooling.
|
||||
|
||||
Package metadata
|
||||
- Package name: idea19_missionledger_verifiable_contract (Python project)
|
||||
- Version: 0.1.0
|
||||
|
||||
This repository is intentionally minimal but designed to evolve into a full production-ready platform.
|
||||
|
|
|
|||
|
|
@ -0,0 +1,7 @@
|
|||
import sys
|
||||
import os
|
||||
|
||||
# Ensure the repository root is on sys.path for test discovery/imports
|
||||
ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__)))
|
||||
if ROOT not in map(os.path.abspath, sys.path):
|
||||
sys.path.insert(0, ROOT)
|
||||
|
|
@ -0,0 +1,23 @@
|
|||
"""MissionLedger: Verifiable, Contract-Driven AI Mission Planner (Core)
|
||||
|
||||
This package provides a tiny, production-ready core to bootstrap the
|
||||
MissionLedger MVP. It includes a lightweight DSL for per-asset LocalProblems
|
||||
and Safety/Privacy contracts, a shadow planner, a simple on-device safety proof
|
||||
hook, and a tamper-evident governance ledger.
|
||||
"""
|
||||
|
||||
from .dsl import LocalProblem, SafetyContracts, PrivacyContracts, PlanDelta
|
||||
from .planner import ShadowPlanner, PlanResult
|
||||
from .safety import verify_plan
|
||||
from .governance import GovernanceLedger
|
||||
|
||||
__all__ = [
|
||||
"LocalProblem",
|
||||
"SafetyContracts",
|
||||
"PrivacyContracts",
|
||||
"PlanDelta",
|
||||
"ShadowPlanner",
|
||||
"PlanResult",
|
||||
"verify_plan",
|
||||
"GovernanceLedger",
|
||||
]
|
||||
|
|
@ -0,0 +1,49 @@
|
|||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass, field
|
||||
from typing import Any, Dict, List
|
||||
|
||||
|
||||
@dataclass
|
||||
class LocalProblem:
|
||||
"""Per-asset planning objective and constraints.
|
||||
|
||||
- assets: mapping from asset_id to asset-specific objectives.
|
||||
- variables: free-form variables used by the planner (e.g., timing, routing).
|
||||
- constraints: list of simple constraint descriptors.
|
||||
- risk_budget: a rough budget to limit risk (abstracted for this MVP).
|
||||
"""
|
||||
|
||||
asset_objectives: Dict[str, Dict[str, Any]] = field(default_factory=dict)
|
||||
variables: Dict[str, Any] = field(default_factory=dict)
|
||||
constraints: List[str] = field(default_factory=list)
|
||||
risk_budget: float = 0.0
|
||||
|
||||
|
||||
@dataclass
|
||||
class SafetyContracts:
|
||||
"""Pre/post conditions and risk terms for safety proofs."""
|
||||
|
||||
pre_conditions: List[str] = field(default_factory=list)
|
||||
post_conditions: List[str] = field(default_factory=list)
|
||||
safety_budget: float = 0.0
|
||||
collision_envelope: float = 0.0
|
||||
cost_of_violation: float = 0.0
|
||||
|
||||
|
||||
@dataclass
|
||||
class PrivacyContracts:
|
||||
"""Minimal data-exposure policies for privacy."""
|
||||
|
||||
allowed_publishers: List[str] = field(default_factory=list)
|
||||
aggregate_signals: bool = True
|
||||
per_message_privacy_budget: float = 0.0
|
||||
|
||||
|
||||
@dataclass
|
||||
class PlanDelta:
|
||||
"""Compact representation of plan changes with metadata for auditability."""
|
||||
|
||||
delta: Dict[str, Any] = field(default_factory=dict)
|
||||
version: int = 0
|
||||
metadata: Dict[str, Any] = field(default_factory=dict)
|
||||
|
|
@ -0,0 +1,38 @@
|
|||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
import json
|
||||
from datetime import datetime
|
||||
from typing import List
|
||||
|
||||
|
||||
class GovernanceLedger:
|
||||
"""A simple tamper-evident governance ledger with per-entry hashes.
|
||||
|
||||
Logs are append-only and anchored via a rolling hash to the previous entry,
|
||||
producing a chain of custody suitable for auditing even on intermittent
|
||||
connectivity.
|
||||
"""
|
||||
|
||||
def __init__(self, device_id: str = "unknown"):
|
||||
self.device_id = device_id
|
||||
self._entries: List[dict] = []
|
||||
self._head_hash = ""
|
||||
|
||||
def log(self, entry: dict) -> None:
|
||||
timestamp = datetime.utcnow().isoformat() + "Z"
|
||||
payload = {
|
||||
"device_id": self.device_id,
|
||||
"timestamp": timestamp,
|
||||
"entry": entry,
|
||||
"parent": self._head_hash,
|
||||
}
|
||||
self._head_hash = hashlib.sha256(json.dumps(payload, sort_keys=True).encode()).hexdigest()
|
||||
self._entries.append(payload)
|
||||
|
||||
@property
|
||||
def entries(self) -> List[dict]:
|
||||
return self._entries[:]
|
||||
|
||||
def latest_hash(self) -> str:
|
||||
return self._head_hash
|
||||
|
|
@ -0,0 +1,38 @@
|
|||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass
|
||||
from typing import Any, Dict
|
||||
|
||||
from .dsl import LocalProblem, PlanDelta
|
||||
from .safety import verify_plan
|
||||
|
||||
|
||||
@dataclass
|
||||
class PlanResult:
|
||||
"""Outcome from planning, including a safe delta and a proof flag."""
|
||||
|
||||
plan_delta: PlanDelta
|
||||
proof_valid: bool
|
||||
justification: str
|
||||
|
||||
|
||||
class ShadowPlanner:
|
||||
"""A lightweight shadow planner that runs in parallel to the executor.
|
||||
|
||||
For this MVP, it generates a conservative delta by slightly increasing
|
||||
resource buffers to create a safe alternative plan if the primary plan is
|
||||
infeasible with respect to LocalProblem.risk_budget.
|
||||
"""
|
||||
|
||||
def __init__(self, local_problem: LocalProblem):
|
||||
self.local_problem = local_problem
|
||||
|
||||
def plan(self, primary_delta: PlanDelta) -> PlanResult:
|
||||
# Very simple heuristic: if risk_budget is low, persevere using a safer delta.
|
||||
risk = float(self.local_problem.risk_budget or 0.0)
|
||||
safe_delta = PlanDelta(delta={"safe_buffer": max(0.0, risk * 0.5)}, version=primary_delta.version + 1, metadata={"owner": "shadow"})
|
||||
|
||||
# Verify safety using the same (mocked) verify_plan hook
|
||||
is_safe = verify_plan(self.local_problem, safe_delta)
|
||||
justification = "Shadow plan generated" if is_safe else "Shadow plan could not be validated safely"
|
||||
return PlanResult(plan_delta=safe_delta, proof_valid=is_safe, justification=justification)
|
||||
|
|
@ -0,0 +1,24 @@
|
|||
from __future__ import annotations
|
||||
|
||||
from typing import Dict
|
||||
|
||||
from .dsl import LocalProblem, PlanDelta
|
||||
|
||||
|
||||
def verify_plan(local_problem: LocalProblem, plan_delta: PlanDelta) -> bool:
|
||||
"""Lightweight safety verification hook.
|
||||
|
||||
This is a minimal, deterministic verifier for the MVP:
|
||||
- If the plan delta carries a non-negative 'safe_buffer', and the local
|
||||
problem's risk_budget is compatible, accept the plan.
|
||||
- Otherwise, reject.
|
||||
This should be replaced with real formal verification hooks in a full build.
|
||||
"""
|
||||
buf = plan_delta.delta.get("safe_buffer", 0.0)
|
||||
try:
|
||||
risk = float(local_problem.risk_budget)
|
||||
except Exception:
|
||||
risk = 0.0
|
||||
|
||||
# Basic rule: safe if buffer does not exceed the risk budget.
|
||||
return float(buf) <= max(0.0, risk)
|
||||
|
|
@ -0,0 +1,15 @@
|
|||
[build-system]
|
||||
requires = ["setuptools>=61.0", "wheel"]
|
||||
build-backend = "setuptools.build_meta"
|
||||
|
||||
[project]
|
||||
name = "idea19_missionledger_verifiable_contract"
|
||||
version = "0.1.0"
|
||||
description = "MVP core for verifiable contract-driven mission planning in intermittently-connected fleets."
|
||||
readme = "README.md"
|
||||
requires-python = ">=3.8"
|
||||
license = {text = "MIT"}
|
||||
|
||||
[tool.setuptools.packages.find]
|
||||
where = ["."]
|
||||
include = ["missionledger*"]
|
||||
|
|
@ -0,0 +1,20 @@
|
|||
"""
|
||||
Site customization for Python startup.
|
||||
|
||||
Ensures the repository root is on sys.path so tests can reliably import
|
||||
the local package (e.g. `import missionledger.dsl`). This is a small,
|
||||
non-invasive shim that helps CI environments where the Python path may not
|
||||
include the project root by default.
|
||||
"""
|
||||
|
||||
import sys
|
||||
import os
|
||||
|
||||
try:
|
||||
# Path to the repository root (directory containing this file)
|
||||
_repo_root = os.path.dirname(os.path.abspath(__file__))
|
||||
if _repo_root not in map(os.path.abspath, sys.path):
|
||||
sys.path.insert(0, _repo_root)
|
||||
except Exception:
|
||||
# Be conservative: do not fail startup if this ever goes wrong
|
||||
pass
|
||||
|
|
@ -0,0 +1,9 @@
|
|||
#!/usr/bin/env bash
|
||||
set -euo pipefail
|
||||
|
||||
echo "==> Running tests with pytest ..."
|
||||
pytest -q
|
||||
|
||||
echo "==> Building the package (python3 -m build) ..."
|
||||
python3 -m build
|
||||
echo "==> Build completed successfully."
|
||||
|
|
@ -0,0 +1,40 @@
|
|||
import json
|
||||
|
||||
from missionledger.dsl import LocalProblem
|
||||
from missionledger.planner import ShadowPlanner
|
||||
from missionledger.dsl import PlanDelta
|
||||
from missionledger.safety import verify_plan
|
||||
from missionledger.governance import GovernanceLedger
|
||||
|
||||
|
||||
def test_shadow_planner_and_safety_verification():
|
||||
# Basic LocalProblem with a modest risk budget
|
||||
lp = LocalProblem(
|
||||
asset_objectives={"rover-1": {"energy": 1.0, "time": 2.0}},
|
||||
variables={"start_time": 0},
|
||||
constraints=["no_overload"],
|
||||
risk_budget=1.0,
|
||||
)
|
||||
|
||||
primary_delta = PlanDelta(delta={"basic_plan": True}, version=1, metadata={})
|
||||
|
||||
planner = ShadowPlanner(lp)
|
||||
shadow = planner.plan(primary_delta)
|
||||
|
||||
# Validate safety proof via the lightweight verifier
|
||||
assert shadow is not None
|
||||
assert isinstance(shadow.plan_delta, PlanDelta)
|
||||
assert shadow.proof_valid in (True, False)
|
||||
|
||||
# Sanity check: the safety verifier should be deterministic for the MVP
|
||||
safe = verify_plan(lp, shadow.plan_delta)
|
||||
assert safe == shadow.proof_valid
|
||||
|
||||
# Governance ledger usage: log an entry and inspect its wiring
|
||||
gl = GovernanceLedger(device_id="test-device")
|
||||
gl.log({"type": "plan_evaluation", "delta_version": shadow.plan_delta.version})
|
||||
assert len(gl.entries) == 1
|
||||
assert "parent" in gl.entries[0]
|
||||
# latest_hash should be a non-empty string hash
|
||||
assert isinstance(gl.latest_hash(), str)
|
||||
assert len(gl.latest_hash()) > 0
|
||||
Loading…
Reference in New Issue