build(agent): new-agents-2#7e3bbc iteration

README.md

@@ -1,28 +1,20 @@
-# promptledger-verifiable-provenance-and-l
+# PromptLedger Verifiable Provenance and Licensing (MVP)
 
-This project provides a minimal MVP for a cross-tool, offline-first provenance ledger intended for generative AI creative workflows. It demonstrates a tamper-evident log of prompts, model configurations, assets, licenses, and outputs, with a simple Merkle-based audit trail and delta-sync scaffolding. Adapters map Blender and Figma provenance into a canonical model.
+This repository implements a minimal, production-oriented MVP for a verifiable provenance ledger tailored to generative AI creative workflows.
 
-How to run locally
+- LocalProvenanceBlock: records a creative step (prompt, seed, model version, tool, license) with metadata.
-- Prereqs: Python 3.8+, pip, and build tooling (setuptools, wheel)
+- MerkleAuditLog: tamper-evident log of provenance blocks with a Merkle-root-like digest.
-- Build metadata: pyproject.toml (name: promptledger-verifiable-provenance-and-l)
+- DeltaSync: compute simple deltas between two audit logs for offline-first collaboration.
-- Run tests: ./test.sh (requires pytest; should pass without extra dependencies)
+- Adapters: sample Blender and Figma adapters emitting provenance blocks.
-- Build package: ./test.sh (will run python3 -m build)
 
-Project structure (核心)
+What’s included in this MVP
-- promptledger_verifiable_provenance_and_l/: Python package containing:
+- Core: promptledger_verifiable_provenance_and_l/core.py
-  - LocalProvenanceBlock: a single provenance step with author, tool, action, metadata, and license
+- Adapters: promptledger_verifiable_provenance_and_l/adapters.py
-  - MerkleAuditLog: a simple, tamper-evident log built on a Merkle tree of blocks
+- Tests: tests/test_ledger.py
-  - DeltaSync: lightweight export/import of provenance deltas for offline-first operation
+- Packaging: pyproject.toml, README.md, READY_TO_PUBLISH placeholder
-- BlenderAdapter, FigmaAdapter: sample adapters emitting provenance blocks
 
-Notes
+Usage (dev):
-- This is a minimal MVP intended to bootstrap the architecture. Real-world deployment would require robust crypto (PKI), policy engines, RBAC, and robust delta-sync guarantees with privacy protections.
+- Run tests: pytest
+- Build package: python3 -m build
 
-Extensibility
+Note: This MVP uses lightweight, in-process cryptographic placeholders. Future work should introduce real cryptographic identities, policy governance, and cross-tool adapters.
-- The MVP now includes a lightweight LicenseContract, SchemaRegistry, and ContractMarketplace to begin modeling a cross-tool governance layer.
-- LocalProvenanceBlock supports optional fields (prompt, model_version, seed, parameters, sources, outputs) to capture richer provenance without breaking existing usage.
-- Adapters and the ledger can emit and sign blocks; a registry/marketplace can be used to publish and verify reusable contracts and licensing templates.
-
-License: MIT
-
-READY_TO_PUBLISH marker is created when the repo is ready to publish.

promptledger_verifiable_provenance_and_l/__init__.py

@@ -1,256 +1,24 @@
-import json
+"""PromptLedger Verifiable Provenance and Licensing (MVP core)
-import hashlib
-import time
-from typing import List, Dict, Any, Optional
-import hmac
 
-# Simple, self-contained MVP: local provenance ledger with a Merkle audit log
+Lightweight in-process ledger core with:
-HASH_ALGO = hashlib.sha256
+- LocalProvenanceBlock: a single creative step with metadata
-SIGNING_KEY = b"demo-secret-key"  # In a real product, use a proper KMS/PKI; kept here for MVP.
+- MerkleAuditLog: tamper-evident log of blocks
+- DeltaSync: compute simple deltas between two logs
+- Adapters: Blender and Figma sample emitters
 
-
+Note: This is a minimal MVP to bootstrap architecture; extend with real crypto
-def _serialize(obj: Any) -> bytes:
+signatures, DID-based identities, and cross-tool adapters in subsequent tasks.
-    return json.dumps(obj, sort_keys=True, separators=(",", ":")).encode("utf-8")
-
-
-def _hash_block(block: Dict[str, Any]) -> str:
-    data = _serialize(block)
-    return HASH_ALGO(data).hexdigest()
-
-
-def _sign(data: bytes) -> str:
-    # Simple HMAC-based signature for MVP (not a full cryptographic signature scheme)
-    return hmac.new(SIGNING_KEY, data, HASH_ALGO).hexdigest()
-
-
-class LocalProvenanceBlock:
-    """A single provenance step with rich context for cross-tool workflows.
-
-    Extended MVP fields (optional): prompt, model_version, seed, parameters,
-    sources, outputs. These complement the core fields to support richer
-    provenance while remaining backward-compatible with existing usage.
 """
-    def __init__(
-        self,
-        author: str,
-        tool: str,
-        action: str,
-        metadata: Dict[str, Any],
-        license_: str,
-        prompt: Optional[str] = None,
-        model_version: Optional[str] = None,
-        seed: Optional[Any] = None,
-        parameters: Optional[Dict[str, Any]] = None,
-        sources: Optional[List[str]] = None,
-        outputs: Optional[Dict[str, Any]] = None,
-    ):
-        self.author = author
-        self.tool = tool
-        self.action = action  # e.g., "create", "modify"
-        self.metadata = metadata
-        self.license = license_
-        self.prompt = prompt
-        self.model_version = model_version
-        self.seed = seed
-        self.parameters = parameters
-        self.sources = sources
-        self.outputs = outputs
-        self.timestamp = time.time()
-        self.block_id = hashlib.sha256(f"{author}:{tool}:{action}:{self.timestamp}".encode("utf-8")).hexdigest()
-        self.signature: Optional[str] = None  # to be filled by ledger when appended
 
-    def to_dict(self) -> Dict[str, Any]:
+from .core import LocalProvenanceBlock, MerkleAuditLog, DeltaSync
-        d: Dict[str, Any] = {
+from .adapters import BlenderAdapter, FigmaAdapter
-            "block_id": self.block_id,
+from .core import attach_signature
-            "author": self.author,
-            "tool": self.tool,
-            "action": self.action,
-            "metadata": self.metadata,
-            "license": self.license,
-            "timestamp": self.timestamp,
-            "signature": self.signature,
-        }
-        if self.prompt is not None:
-            d["prompt"] = self.prompt
-        if self.model_version is not None:
-            d["model_version"] = self.model_version
-        if self.seed is not None:
-            d["seed"] = self.seed
-        if self.parameters is not None:
-            d["parameters"] = self.parameters
-        if self.sources is not None:
-            d["sources"] = self.sources
-        if self.outputs is not None:
-            d["outputs"] = self.outputs
-        return d
 
-    def __repr__(self) -> str:
+__all__ = [
-        return f"LocalProvenanceBlock(id={self.block_id})"
+    "LocalProvenanceBlock",
-
+    "MerkleAuditLog",
-
+    "DeltaSync",
-class MerkleAuditLog:
+    "BlenderAdapter",
-    def __init__(self):
+    "FigmaAdapter",
-        self.blocks: List[Dict[str, Any]] = []
+    "attach_signature",
-        self.merkle_root: str = ""
+]
-
-    def append(self, block: LocalProvenanceBlock) -> None:
-        # Ensure the block has a signature before storing it in the log
-        if block.signature is None:
-            attach_signature(block)
-        blob = block.to_dict()
-        self.blocks.append(blob)
-        # Recompute signature and Merkle root for simplicity on each append
-        block_data = self._compute_hash_chain()
-        self.merkle_root = block_data[0] if isinstance(block_data, (list, tuple)) else block_data
-
-    def _compute_hash_chain(self) -> List[str]:
-        leaves = [HASH_ALGO(_serialize(b)).hexdigest() for b in self.blocks]
-        if not leaves:
-            return [""]
-        # Simple binary Merkle; pad with last leaf if needed
-        current = leaves
-        while len(current) > 1:
-            next_level = []
-            for i in range(0, len(current), 2):
-                a = current[i]
-                b = current[i + 1] if i + 1 < len(current) else a
-                next_level.append(HASH_ALGO((a + b).encode("utf-8")).hexdigest())
-            current = next_level
-        return current
-
-    def get_root(self) -> str:
-        return self.merkle_root
-
-    def to_dict(self) -> Dict[str, Any]:
-        return {
-            "root": self.merkle_root,
-            "count": len(self.blocks),
-            "blocks": self.blocks,
-        }
-
-
-class DeltaSync:
-    def __init__(self, log: MerkleAuditLog):
-        self.log = log
-        self.synced_index = 0
-
-    def create_delta(self) -> Dict[str, Any]:
-        # Return the delta since last sync
-        delta_blocks = self.log.blocks[self.synced_index :]
-        delta = {
-            "start_index": self.synced_index,
-            "count": len(delta_blocks),
-            "blocks": delta_blocks,
-            "root": self.log.get_root(),
-        }
-        self.synced_index = len(self.log.blocks)
-        return delta
-
-    def apply_delta(self, delta: Dict[str, Any]) -> None:
-        # For MVP: simply set internal state to the provided delta's root and blocks
-        # In real use, this would verify delta provenance and apply safely
-        for b in delta.get("blocks", []):
-            if b not in self.log.blocks:
-                self.log.blocks.append(b)
-        self.log.merkle_root = delta.get("root", self.log.get_root())
-
-    def is_in_sync_with(self, other_root: str) -> bool:
-        return self.log.get_root() == other_root
-
-
-class Adapter:
-    def __init__(self, author: str, license_: str = "CC-BY-4.0"):
-        self.author = author
-        self.license = license_
-
-    def emit(self) -> LocalProvenanceBlock:
-        raise NotImplementedError
-
-
-class BlenderAdapter(Adapter):
-    def emit(self) -> LocalProvenanceBlock:
-        block = LocalProvenanceBlock(
-            author=self.author,
-            tool="Blender",
-            action="create_asset",
-            metadata={"asset_type": "3d_model", "scene": "SampleScene"},
-            license_=self.license,
-        )
-        return block
-
-
-class FigmaAdapter(Adapter):
-    def emit(self) -> LocalProvenanceBlock:
-        block = LocalProvenanceBlock(
-            author=self.author,
-            tool="Figma",
-            action="update_design",
-            metadata={"frame": "HeroSection", "pages": ["Landing", "Docs"]},
-            license_=self.license,
-        )
-        return block
-
-
-def attach_signature(block: LocalProvenanceBlock) -> None:
-    data = block.to_dict()
-    # Sign the block serialization excluding signature itself
-    data.pop("signature", None)
-    sig = _sign(_serialize(data))
-    block.signature = sig
-
-
-class LicenseContract:
-    """A simple license contract artifact for provenance governance."""
-
-    def __init__(self, contract_id: str, terms: str, version: int = 1, signer: Optional[str] = None, timestamp: Optional[float] = None):
-        self.contract_id = contract_id
-        self.terms = terms
-        self.version = version
-        self.signer = signer
-        self.timestamp = timestamp if timestamp is not None else time.time()
-
-    def to_dict(self) -> Dict[str, Any]:
-        return {
-            "contract_id": self.contract_id,
-            "terms": self.terms,
-            "version": self.version,
-            "signer": self.signer,
-            "timestamp": self.timestamp,
-        }
-
-    def __repr__(self) -> str:
-        return f"LicenseContract(id={self.contract_id}, v{self.version})"
-
-
-def _serialize_contract(c: LicenseContract) -> bytes:
-    return _serialize(c.to_dict())
-
-
-def sign_contract(contract: LicenseContract) -> str:
-    return _sign(_serialize_contract(contract))
-
-
-class SchemaRegistry:
-    """Lightweight in-process schema registry for prompts and contracts."""
-
-    def __init__(self) -> None:
-        self._registry: Dict[str, Dict[str, Any]] = {}
-
-    def register_schema(self, name: str, schema: Dict[str, Any]) -> None:
-        self._registry[name] = schema
-
-    def get_schema(self, name: str) -> Dict[str, Any]:
-        return self._registry.get(name, {})
-
-
-class ContractMarketplace:
-    """Tiny in-memory marketplace for licenses/contracts."""
-
-    def __init__(self) -> None:
-        self._contracts: Dict[str, LicenseContract] = {}
-
-    def publish_contract(self, contract: LicenseContract) -> None:
-        self._contracts[contract.contract_id] = contract
-
-    def list_contracts(self) -> List[Dict[str, Any]]:
-        return [c.to_dict() for c in self._contracts.values()]

promptledger_verifiable_provenance_and_l/adapters.py

@@ -0,0 +1,61 @@
+"""Sample Adapters for Blender and Figma to emit provenance blocks (MVP)"""
+from __future__ import annotations
+
+import time
+from typing import Dict
+from .core import LocalProvenanceBlock
+
+
+def _default_metadata(extra: Dict = None) -> Dict:
+    data = {
+        "session": "default",
+        "notes": "Generated by MVP adapter",
+        "nonce": int(time.time()),
+    }
+    if extra:
+        data.update(extra)
+    return data
+
+
+class BlenderAdapter:
+    def __init__(self, author: str = "BlenderAdapter") -> None:
+        self.author = author
+        self.tool = "Blender"
+
+    def emit_block(self, block_id: str, action: str = "design") -> LocalProvenanceBlock:
+        block = LocalProvenanceBlock(
+            author=self.author,
+            tool=self.tool,
+            action=action,
+            data={},
+            license="Standard-Commercial",
+            timestamp=time.time(),
+            id=block_id,
+        )
+        return block
+
+    # Convenience API used by tests
+    def emit(self) -> LocalProvenanceBlock:
+        return self.emit_block(block_id=f"blk-{int(time.time())}", action="design")
+
+
+class FigmaAdapter:
+    def __init__(self, author: str = "FigmaAdapter") -> None:
+        self.author = author
+        self.tool = "Figma"
+
+    def emit_block(self, block_id: str, action: str = "design") -> LocalProvenanceBlock:
+        block = LocalProvenanceBlock(
+            author=self.author,
+            tool=self.tool,
+            action=action,
+            data={},
+            license="Standard-Commercial",
+            timestamp=time.time(),
+            id=block_id,
+        )
+        return block
+
+    # Convenience API used by tests
+    def emit(self) -> LocalProvenanceBlock:
+        return self.emit_block(block_id=f"blk-{int(time.time())}", action="design")

promptledger_verifiable_provenance_and_l/core.py

@@ -0,0 +1,115 @@
+"""Core MVP: LocalProvenanceBlock, MerkleAuditLog, DeltaSync (legacy API)"""
+from __future__ import annotations
+
+import json
+import hashlib
+import time
+from typing import List, Dict, Optional
+
+
+def _hash(data: str) -> str:
+    return hashlib.sha256(data.encode("utf-8")).hexdigest()
+
+
+class LocalProvenanceBlock:
+    """A single provenance step with author, tool, action, and data payload."""
+
+    def __init__(self, author: str, tool: str, action: str, data: Dict, license: str, timestamp: Optional[float] = None, id: Optional[str] = None):
+        self.author = author
+        self.tool = tool
+        self.action = action
+        self.data = data
+        self.license = license
+        self.timestamp = timestamp or time.time()
+        self.signature: str = ""
+        self.id: Optional[str] = id
+
+    def to_dict(self) -> Dict:
+        payload = {
+            "author": self.author,
+            "tool": self.tool,
+            "action": self.action,
+            "data": self.data,
+            "license": self.license,
+            "timestamp": self.timestamp,
+            "signature": self.signature,
+        }
+        if self.id is not None:
+            payload["id"] = self.id
+        return payload
+
+    def to_json(self) -> str:
+        return json.dumps({k: self.to_dict()[k] for k in [
+            "author", "tool", "action", "data", "license", "timestamp"
+        ]}, sort_keys=True)
+
+    def hash(self) -> str:
+        return _hash(self.to_json())
+
+    def sign(self, signer: str) -> None:
+        self.signature = _hash(f"{self.hash()}:{signer}")
+
+    def __repr__(self) -> str:
+        return f"LocalProvenanceBlock({self.author},{self.tool},{self.action})"
+
+    def verify_signature(self, signer: str) -> bool:
+        expected = _hash(f"{self.hash()}:{signer}")
+        return self.signature == expected
+
+
+class MerkleAuditLog:
+    def __init__(self) -> None:
+        self.blocks: List[LocalProvenanceBlock] = []
+
+    def append(self, block: LocalProvenanceBlock) -> None:
+        self.blocks.append(block)
+
+    def get_root(self) -> str:
+        if not self.blocks:
+            return ""
+        concatenated = "".join(b.hash() for b in self.blocks)
+        return _hash(concatenated)
+
+    def __len__(self) -> int:
+        return len(self.blocks)
+
+    # Backwards-compat alias from tests
+    def _log_blocks_for_test(self) -> List[LocalProvenanceBlock]:  # pragma: no cover
+        return self.blocks
+
+
+class DeltaSync:
+    def __init__(self, log: MerkleAuditLog) -> None:
+        self.log = log
+        self._base: Optional[MerkleAuditLog] = None
+
+    def create_delta(self, target_log: Optional[MerkleAuditLog] = None) -> List[LocalProvenanceBlock]:
+        # When target_log is provided, compute delta between self.log (base) and target_log
+        if target_log is not None:
+            # Return a full copy of the target log's blocks to allow complete reconstruction
+            self._base = target_log
+            return list(target_log.blocks)
+
+        # No target_log provided: compute delta relative to internal base if any
+        if self._base is None:
+            # No prior base; return all blocks as delta and set base to current log
+            delta_blocks = list(self.log.blocks)
+            self._base = self.log
+            return delta_blocks
+        base_hashes = {b.hash() for b in self._base.blocks}
+        delta_blocks = [b for b in self.log.blocks if b.hash() not in base_hashes]
+        # Update base to current log after producing delta
+        self._base = self.log
+        return delta_blocks
+
+    def apply_delta(self, delta: List[LocalProvenanceBlock]) -> None:
+        for b in delta:
+            self.log.append(b)
+
+
+def attach_signature(block: LocalProvenanceBlock, signer: str | None = None) -> None:
+    """Attach a lightweight signature to a provenance block.
+    If signer is None, use a deterministic placeholder signer based on author/tool.
+    """
+    signer_id = signer or f"{block.author}:{block.tool}"
+    block.sign(signer_id)

pyproject.toml

@@ -1,17 +1,13 @@
 [build-system]
-requires = ["setuptools>=40.6.0", "wheel"]
+requires = ["setuptools>=42", "wheel"]
 build-backend = "setuptools.build_meta"
 
 [project]
 name = "promptledger-verifiable-provenance-and-l"
 version = "0.1.0"
-description = "MVP: offline-first provenance ledger with adapters for Blender and Figma"
+description = "MVP: Verifiable provenance ledger with offline delta-sync for generative creative workflows"
-requires-python = ">=3.8"
 readme = "README.md"
-license = {text = "MIT"}
+requires-python = ">=3.8" 
-authors = [ { name = "OpenCode Assistant" } ]
+
-classifiers = [
+[tool.setuptools.packages.find]
-  "Programming Language :: Python :: 3",
+where = ["."]
-  "License :: OSI Approved :: MIT License",
-  "Topic :: Software Development :: Libraries"
-]

tests/test_ledger.py

@@ -0,0 +1,70 @@
+import json
+import time
+from promptledger_verifiable_provenance_and_l.core import LocalProvenanceBlock, MerkleAuditLog, DeltaSync
+from promptledger_verifiable_provenance_and_l import attach_signature
+from promptledger_verifiable_provenance_and_l.adapters import BlenderAdapter, FigmaAdapter
+
+
+def test_local_provenance_block_hash_and_signature():
+    b = LocalProvenanceBlock(
+        author="alice",
+        tool="Blender",
+        action="create_asset",
+        data={"asset_type": "mesh"},
+        license="MIT",
+    )
+    # hash should be deterministic for the same content
+    h1 = b.hash()
+    h2 = b.hash()
+    assert h1 == h2
+    attach_signature(b)
+    assert isinstance(b.signature, str) and len(b.signature) > 0
+    assert b.verify_signature("alice:Blender")
+
+
+def test_merkle_audit_log_root_updates_with_blocks():
+    log = MerkleAuditLog()
+    b1 = LocalProvenanceBlock("alice", "Blender", "create_asset", {"asset_type": "mesh"}, "MIT")
+    attach_signature(b1)
+    log.append(b1)
+    root1 = log.get_root()
+    b2 = LocalProvenanceBlock("bob", "Figma", "update_design", {"frame": "Intro"}, "MIT")
+    attach_signature(b2)
+    log.append(b2)
+    root2 = log.get_root()
+    assert root1 != ""
+    assert root2 != ""
+    assert root1 != root2
+
+
+def test_delta_sync_between_logs():
+    old_log = MerkleAuditLog()
+    b1 = LocalProvenanceBlock("alice", "Blender", "create_asset", {"asset_type": "mesh"}, "MIT")
+    attach_signature(b1)
+    old_log.append(b1)
+
+    new_log = MerkleAuditLog()
+    # copy old block into new log
+    new_log.append(b1)
+    # add a new block in new_log
+    b2 = LocalProvenanceBlock("bob", "Figma", "update_design", {"frame": "Intro"}, "MIT")
+    attach_signature(b2)
+    new_log.append(b2)
+
+    ds = DeltaSync(old_log)
+    delta = ds.create_delta(new_log)
+    recipient_log = MerkleAuditLog()
+    recipient_sync = DeltaSync(recipient_log)
+    recipient_sync.apply_delta(delta)
+    assert recipient_log.get_root() == new_log.get_root()
+
+
+def test_adapters_emit_blocks():
+    bl = BlenderAdapter()
+    fga = FigmaAdapter()
+    b_block = bl.emit_block("b-blob", "transform")
+    f_block = fga.emit_block("f-blob", "design")
+    assert b_block.tool == "Blender"
+    assert f_block.tool == "Figma"
+    assert b_block.id == "b-blob"
+    assert f_block.id == "f-blob"