build(agent): new-agents-3#dd492b iteration

README.md

@@ -18,6 +18,11 @@ Project structure (核心)
 Notes
 - 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.
 
+Extensibility
+- 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,7 +1,7 @@
 import json
 import hashlib
 import time
-from typing import List, Dict, Any
+from typing import List, Dict, Any, Optional
 import hmac
 
 # Simple, self-contained MVP: local provenance ledger with a Merkle audit log
@@ -24,18 +24,43 @@ def _sign(data: bytes) -> str:
 
 
 class LocalProvenanceBlock:
-    def __init__(self, author: str, tool: str, action: str, metadata: Dict[str, Any], license_: str):
+    """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 = None  # to be filled by ledger when appended
+        self.signature: str | None = None  # to be filled by ledger when appended
 
     def to_dict(self) -> Dict[str, Any]:
-        return {
+        d: Dict[str, Any] = {
             "block_id": self.block_id,
             "author": self.author,
             "tool": self.tool,
@@ -45,6 +70,19 @@ class LocalProvenanceBlock:
             "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:
         return f"LocalProvenanceBlock(id={self.block_id})"
@@ -157,3 +195,60 @@ def attach_signature(block: LocalProvenanceBlock) -> None:
     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()]