Download questionos/underwrite.py from josephrw/systemlake-underwriter: direct link, hf CLI and curl.
- Browser
- Download file 13.3 kB
-
https://huggingface.co/josephrw/systemlake-underwriter/resolve/main/questionos/underwrite.py
- Command line
-
hf download hf://josephrw/systemlake-underwriter/questionos/underwrite.py
-
curl -L -o underwrite.py https://huggingface.co/josephrw/systemlake-underwriter/resolve/main/questionos/underwrite.py
13.3 kB
| """Unified Value Claim Packet — questionos.underwrite | |
| Combines SystemLake crawl + UnderwritingEngine + AAU into one command | |
| that produces a Value Claim Packet from a repo or local system. | |
| Usage: | |
| python3 -m questionos.underwrite /path/to/repo --out vcp.json --b64 | |
| The packet contains: | |
| - System lake summary (Merkle root, file counts, system detection) | |
| - Collateral scores (functionality, deployability, receipt strength, etc.) | |
| - Adversarial attribution underwriting (gaming checks, counterfactual stripping) | |
| - Value claim with settlement status | |
| - Cognition packet (redacted, compressed, Base64-ready) | |
| - Receipt chain proving the underwriting occurred | |
| The packet does NOT contain: | |
| - Raw file contents | |
| - Secrets or tokens | |
| - Sensitive zone metadata | |
| """ | |
| import os | |
| import sys | |
| import json | |
| import hashlib | |
| import base64 | |
| import zlib | |
| import argparse | |
| import tempfile | |
| from datetime import datetime | |
| from typing import Dict, List, Optional | |
| from systemlake.lake import MachineLake | |
| from systemlake.policy import PolicyEngine, RedactionEngine | |
| from systemlake.compressor import CognitionCompressor | |
| from systemlake.underwriter import UnderwritingEngine, CollateralScore | |
| from systemlake.aau import ( | |
| AdversarialAttributionUnderwriter, | |
| Baseline, Evidence, GamingFlags, ValueClaim, ClaimStatus, | |
| ) | |
| from quadrantos.receipt_store import SQLiteReceiptStore | |
| def generate_value_claim_packet( | |
| repo_path: str, | |
| output_path: str = None, | |
| emit_b64: bool = False, | |
| lake_db: str = None, | |
| receipts_db: str = None, | |
| ) -> Dict: | |
| """Generate a Value Claim Packet from a repo or local system. | |
| This is the unified command that ties together: | |
| 1. SystemLake crawl → SQLite index, Merkle root, system detection | |
| 2. UnderwritingEngine → collateral scores (7 dimensions + haircuts) | |
| 3. AAU → adversarial attribution, gaming detection, settlement | |
| 4. CognitionCompressor → redacted semantic packet | |
| 5. SQLiteReceiptStore → receipt chain proving the underwriting | |
| Args: | |
| repo_path: Path to the repo/system to underwrite | |
| output_path: Where to write the VCP JSON (default: stdout) | |
| emit_b64: Also emit a Base64 compressed version | |
| lake_db: Path to the lake SQLite DB (default: temp) | |
| receipts_db: Path to the receipts SQLite DB (default: temp) | |
| Returns: | |
| The Value Claim Packet dict | |
| """ | |
| repo_path = os.path.expanduser(repo_path) | |
| if not os.path.isdir(repo_path): | |
| raise ValueError(f"Path does not exist: {repo_path}") | |
| now_str = datetime.now().isoformat() | |
| tmp_dir = tempfile.mkdtemp(prefix='vcp_') | |
| lake_db = lake_db or os.path.join(tmp_dir, 'lake.db') | |
| receipts_db = receipts_db or os.path.join(tmp_dir, 'receipts.db') | |
| # --- Layer 1: MachineLake Crawl --- | |
| lake = MachineLake(db_path=lake_db) | |
| crawl_result = lake.crawl(repo_path, max_files=1000) | |
| # --- Layer 2: Collateral Underwriting --- | |
| underwriter = UnderwritingEngine(lake_db) | |
| collateral_scores = underwriter.score_all() | |
| # --- Layer 3: Adversarial Attribution Underwriting --- | |
| aau = AdversarialAttributionUnderwriter() | |
| aau_results = [] | |
| for score in collateral_scores: | |
| # Build baseline from collateral dimensions | |
| baseline = Baseline( | |
| label=f"{score.system_name}_baseline", | |
| snapshot_hash=crawl_result['merkle_root'], | |
| timestamp=now_str, | |
| metrics={ | |
| 'functionality': score.functionality, | |
| 'deployability': score.deployability, | |
| 'receipt_strength': score.receipt_strength, | |
| 'security': score.security_cleanliness, | |
| }, | |
| ) | |
| # Build evidence from detected capabilities | |
| evidence_list = [] | |
| systems = lake.list_systems() | |
| for sys_row in systems: | |
| if sys_row['name'] == score.system_name: | |
| if sys_row['has_receipts']: | |
| evidence_list.append(Evidence( | |
| evidence_id=hashlib.sha256( | |
| f"{score.system_name}_receipts".encode()).hexdigest()[:16], | |
| kind='receipt', | |
| source=f"{score.system_name}/receipts", | |
| timestamp=now_str, | |
| payload_hash=crawl_result['merkle_root'][:16], | |
| weight=0.8, | |
| verified=True, | |
| )) | |
| if sys_row['has_tests']: | |
| evidence_list.append(Evidence( | |
| evidence_id=hashlib.sha256( | |
| f"{score.system_name}_tests".encode()).hexdigest()[:16], | |
| kind='test_pass', | |
| source=f"{score.system_name}/tests", | |
| timestamp=now_str, | |
| payload_hash='test_detected', | |
| weight=0.7, | |
| verified=True, | |
| )) | |
| if sys_row['has_endpoints']: | |
| evidence_list.append(Evidence( | |
| evidence_id=hashlib.sha256( | |
| f"{score.system_name}_endpoints".encode()).hexdigest()[:16], | |
| kind='endpoint_response', | |
| source=f"{score.system_name}/endpoints", | |
| timestamp=now_str, | |
| payload_hash='endpoint_detected', | |
| weight=0.6, | |
| verified=False, # not actually called | |
| )) | |
| break | |
| # Detect gaming flags from collateral score | |
| gaming = GamingFlags() | |
| if score.secret_risk > 0: | |
| gaming.unverifiable_delta = False # secrets don't mean no delta | |
| if score.missing_tests > 0: | |
| gaming.unverifiable_delta = True | |
| if score.unverifiable_claims > 0: | |
| gaming.unverifiable_delta = True | |
| if score.no_users > 0: | |
| gaming.non_poolable = True | |
| if score.collateral_score < 20: | |
| gaming.non_poolable = True | |
| # Estimate value (conservative) | |
| # Base: hours of work represented × rate × confidence | |
| estimated_hours = min(score.functionality / 10, 20) # cap at 20h | |
| estimated_value = estimated_hours * 120 * (score.collateral_score / 100) | |
| counterfactual = estimated_value * 0.3 # 30% would have happened anyway | |
| claim = ValueClaim( | |
| claim_id=hashlib.sha256( | |
| f"{score.system_name}_{now_str}".encode()).hexdigest()[:16], | |
| system_name=score.system_name, | |
| baseline=baseline, | |
| evidence=evidence_list, | |
| gaming_flags=gaming, | |
| claimed_value_usd=estimated_value, | |
| counterfactual_value_usd=counterfactual, | |
| hours_avoided=estimated_hours, | |
| confidence=score.collateral_score / 100, | |
| exchangeability=min(1.0, score.deployability / 100), | |
| reputation=0.5, # neutral start | |
| ) | |
| aau_result = aau.underwrite(claim) | |
| aau_results.append({ | |
| 'system_name': score.system_name, | |
| 'collateral_score': score.to_dict(), | |
| 'aau_receipt': aau_result['receipt'], | |
| 'status': aau_result['status'].value, | |
| 'settled_value': aau_result['settled_value'], | |
| 'gaming_flags': aau_result['gaming_flags'], | |
| }) | |
| # --- Layer 4: Cognition Packet --- | |
| policy = PolicyEngine() | |
| redactor = RedactionEngine() | |
| compressor = CognitionCompressor(lake, policy, redactor) | |
| cognition_packet = compressor.compress( | |
| root=repo_path, max_files=200, include_snippets=True, include_symbols=True) | |
| cognition_receipt = compressor.to_receipt(cognition_packet) | |
| # --- Layer 5: Receipt Chain --- | |
| receipts = SQLiteReceiptStore(receipts_db) | |
| vcp_receipt = receipts.write( | |
| agent='ValueClaimPacket', | |
| action='vcp_generated', | |
| artifact_path=output_path, | |
| details={ | |
| 'repo': repo_path, | |
| 'merkle_root': crawl_result['merkle_root'][:16], | |
| 'systems_scored': len(collateral_scores), | |
| 'aau_settlements': len(aau_results), | |
| 'cognition_packet_hash': cognition_receipt['packet_sha256'][:16], | |
| }, | |
| ) | |
| # --- Assemble VCP --- | |
| vcp = { | |
| 'schema': 'membra.value_claim_packet.v1', | |
| 'generated_at': now_str, | |
| 'repo_path': repo_path, | |
| 'repo_name': os.path.basename(repo_path), | |
| 'lake': { | |
| 'merkle_root': crawl_result['merkle_root'], | |
| 'file_count': crawl_result['file_count'], | |
| 'total_size': crawl_result['total_size'], | |
| 'new_files': crawl_result['new_files'], | |
| 'changed_files': crawl_result['changed_files'], | |
| 'crawl_duration_ms': crawl_result['duration_ms'], | |
| }, | |
| 'systems': [s.to_dict() for s in collateral_scores], | |
| 'underwriting': aau_results, | |
| 'cognition': { | |
| 'packet_sha256': cognition_receipt['packet_sha256'], | |
| 'b64_size': cognition_receipt['b64_size'], | |
| 'privacy': cognition_packet['privacy'], | |
| 'systems_detected': len(cognition_packet.get('systems', [])), | |
| }, | |
| 'settlement_chain': aau.verify_settlements(), | |
| 'receipt': { | |
| 'id': vcp_receipt['id'], | |
| 'chain_hash': vcp_receipt['chain_hash'], | |
| 'timestamp': vcp_receipt['timestamp'], | |
| }, | |
| 'disclaimer': ( | |
| 'Estimated underwritten software work-equity, not guaranteed valuation. ' | |
| 'Every claim has been through adversarial attribution review. ' | |
| 'Settled values are finance-readable estimates subject to external audit. ' | |
| 'Collateral scores are not cash net worth.' | |
| ), | |
| } | |
| # Compute VCP hash | |
| vcp['vcp_sha256'] = hashlib.sha256( | |
| json.dumps(vcp, sort_keys=True).encode()).hexdigest() | |
| # --- Output --- | |
| if output_path: | |
| with open(output_path, 'w') as f: | |
| json.dump(vcp, f, indent=2) | |
| if emit_b64: | |
| raw = json.dumps(vcp, sort_keys=True).encode() | |
| compressed = zlib.compress(raw, 9) | |
| b64 = base64.b64encode(compressed).decode() | |
| b64_path = (output_path or '/dev/stdout').replace('.json', '.b64') | |
| if output_path: | |
| with open(b64_path, 'w') as f: | |
| f.write(b64) | |
| vcp['b64_path'] = b64_path | |
| vcp['b64_size'] = len(b64) | |
| # Cleanup temp | |
| import shutil | |
| shutil.rmtree(tmp_dir, ignore_errors=True) | |
| return vcp | |
| def print_report(vcp: Dict): | |
| """Print a human-readable VCP report.""" | |
| print("=" * 70) | |
| print(" MEMBRA VALUE CLAIM PACKET") | |
| print("=" * 70) | |
| print() | |
| print(f" Repo: {vcp['repo_name']}") | |
| print(f" Merkle root: {vcp['lake']['merkle_root'][:16]}") | |
| print(f" Files indexed: {vcp['lake']['file_count']}") | |
| print(f" Systems scored: {len(vcp['systems'])}") | |
| print() | |
| print(" COLLATERAL SCORES:") | |
| for s in vcp['systems']: | |
| print(f" {s['system_name']:30s} Score: {s['collateral_score']:5.1f} Grade: {s['grade']}") | |
| print() | |
| print(" ADVERSARIAL ATTRIBUTION UNDERWRITING:") | |
| for u in vcp['underwriting']: | |
| r = u['aau_receipt'] | |
| print(f" {u['system_name']:30s} Status: {u['status']}") | |
| print(f" Claimed: ${r['claimed_value_usd']:.2f} " | |
| f"Counterfactual: ${r['counterfactual_value_usd']:.2f} " | |
| f"Net: ${r['net_value_usd']:.2f}") | |
| print(f" Settled: ${u['settled_value']:.2f} " | |
| f"Score: {r['final_score']:.1f}") | |
| if u['gaming_flags']: | |
| print(f" Gaming flags: {', '.join(u['gaming_flags'])}") | |
| print() | |
| print(f" Settlement chain: {vcp['settlement_chain']}") | |
| print(f" VCP SHA-256: {vcp['vcp_sha256'][:16]}") | |
| print(f" Receipt: {vcp['receipt']['id'][:8]} chain={vcp['receipt']['chain_hash'][:16]}") | |
| print() | |
| print(f" {vcp['disclaimer']}") | |
| print("=" * 70) | |
| def main(): | |
| """CLI entry point: python3 -m questionos.underwrite /path/to/repo --out vcp.json --b64""" | |
| ap = argparse.ArgumentParser( | |
| description='Generate a Value Claim Packet from a repo or local system.') | |
| ap.add_argument('repo', help='Path to the repo/system to underwrite') | |
| ap.add_argument('--out', default=None, help='Output JSON file path') | |
| ap.add_argument('--b64', action='store_true', help='Also emit Base64 compressed version') | |
| ap.add_argument('--lake-db', default=None, help='Path to lake SQLite DB') | |
| ap.add_argument('--receipts-db', default=None, help='Path to receipts SQLite DB') | |
| ap.add_argument('--quiet', action='store_true', help='Suppress report output') | |
| args = ap.parse_args() | |
| vcp = generate_value_claim_packet( | |
| repo_path=args.repo, | |
| output_path=args.out, | |
| emit_b64=args.b64, | |
| lake_db=args.lake_db, | |
| receipts_db=args.receipts_db, | |
| ) | |
| if not args.quiet: | |
| print_report(vcp) | |
| if args.out: | |
| print(f"\n VCP written to: {args.out}") | |
| if args.b64 and 'b64_path' in vcp: | |
| print(f" B64 written to: {vcp['b64_path']} ({vcp['b64_size']} chars)") | |
| if __name__ == '__main__': | |
| main() | |