"""Offline Joey Nguyen (model 002) production gates. Never calls a provider or starts a paid job. Usage: python production-helper.py cache-key request.json python production-helper.py plan request.json --config production.json --ledger production-ledger.json python production-helper.py reserve request.json --config production.json --ledger production-ledger.json python production-helper.py summarize --ledger production-ledger.json After a real provider job, update its ledger entry with provider job ID, real output hash, QA evidence and all actual cost components. Unknown costs remain null. A reservation is not a completed render. """ import argparse import hashlib import json import re from pathlib import Path SHA256 = re.compile(r"^[0-9a-f]{64}$") LIVE_STATUSES = {"reserved", "submitted", "rendered", "accepted", "rejected", "failed", "discarded"} def cache_key(request): renderer = request.get("renderer", {}) payload = { "stage": request.get("stage"), "prompt": request.get("prompt"), "reference_sha256": request.get("reference_sha256", []), "first_frame_sha256": request.get("first_frame", {}).get("sha256"), "brief": request.get("brief", {}), "aspect_ratio": request.get("brief", {}).get("aspect_ratio"), "renderer": {key: renderer.get(key) for key in ( "provider", "model_id", "model_version", "settings", "resolution", "audio_features" )}, } canonical = json.dumps(payload, sort_keys=True, separators=(",", ":"), ensure_ascii=False) return hashlib.sha256(canonical.encode()).hexdigest() def shot_signature(request): payload = {"brief": request.get("brief", {}), "prompt": request.get("prompt"), "reference_sha256": request.get("reference_sha256", []), "first_frame_sha256": request.get("first_frame", {}).get("sha256")} return hashlib.sha256(json.dumps(payload, sort_keys=True, separators=(",", ":")).encode()).hexdigest() def summarize(ledger): entries = ledger.get("entries", []) known, committed, unknown = 0, 0, 0 accepted = set() for entry in entries: components = entry.get("cost_components", []) if entry.get("status") not in {"reserved"}: entry_unknown = not components for component in components: amount = component.get("usd_micros") if amount is None: entry_unknown = True else: if not isinstance(amount, int) or isinstance(amount, bool) or amount < 0: raise ValueError("Costs must be nonnegative integer USD micros or null.") known += amount unknown += int(entry_unknown) external = sum( c["usd_micros"] for c in components if c.get("external_api") and isinstance(c.get("usd_micros"), int) ) if entry.get("status") in LIVE_STATUSES: quoted = entry.get("estimated_external_cost_usd_micros") if entry.get("cost_finalized") is True: if not components: raise ValueError("Finalized cost needs actual cost components.") if any(c.get("external_api") and c.get("usd_micros") is None for c in components): raise ValueError("Finalized external cost cannot remain unknown.") committed += external else: if not isinstance(quoted, int) or isinstance(quoted, bool) or quoted < 0: raise ValueError("Every unfinalized API job needs a reserved cost upper bound.") committed += max(quoted, external) if entry.get("status") == "accepted": accepted.add(entry.get("deliverable_id") or entry["shot_id"]) return { "status": "no_jobs_recorded" if not entries else "partial_cost" if unknown else "recorded_cost", "known_total_usd_micros": known, "external_committed_usd_micros": committed, "unknown_cost_entries": unknown, "accepted_deliverables": len(accepted), "cost_per_accepted_deliverable_usd_micros": round(known / len(accepted)) if accepted and unknown == 0 else None, } def verify_short_voice_review(report, output_sha256): """Validate machine-review evidence for one exact stored speech clip. This gate does not claim human listening or approve an untested future clip. """ try: voice = report["voice"] sync = voice["syncnet"] actual = sync["actual"] crop = sync["crop"] asr = voice["asr"] waveform = voice["waveform"] required_gates = { "official_example_matches_expected_offset_3", "known_delayed_audio_recovered", "known_advanced_audio_recovered", "absolute_actual_offset_within_80ms", "actual_confidence_above_3", "reversed_audio_fails_sync_gate", } if not (SHA256.fullmatch(output_sha256 or "") and voice["output_sha256"] == output_sha256 == sync["source_video_sha256"] and voice["source_audio_sha256"] == asr["master_audio_sha256"] and voice["approved_scope"] == "exact_stored_short_voice_proof" and required_gates == set(sync["project_gates"]) and all(value is True for value in sync["project_gates"].values()) and abs(actual["offset_seconds"]) <= 0.08 and actual["confidence"] > 3 and actual["analysis_feature_windows"] > 0 and actual["coverage_seconds_from_first_to_last_window_end"] >= voice["source_audio_duration_seconds"] - 0.2 and crop["missing_detections"] == 0 and crop["face_detections"] == crop["frames_25fps"] > 0 and waveform["waveform_correlation"] >= 0.99 and abs(waveform["candidate_offset_seconds"]) <= 0.08 and waveform["overlap_seconds"] >= voice["source_audio_duration_seconds"] - 0.05 and sync["official_example"]["offset_frames"] == 3): return False controls = sync["shift_controls"] if set(controls) != {"audio_delayed_480ms", "audio_advanced_480ms"}: return False if controls["audio_delayed_480ms"]["known_audio_shift_seconds"] != 0.48 or controls["audio_advanced_480ms"]["known_audio_shift_seconds"] != -0.48: return False for control in controls.values(): expected_offset = actual["offset_seconds"] - control["known_audio_shift_seconds"] if abs(control["offset_seconds"] - expected_offset) > 0.040001: return False negative = sync["mismatched_audio_negative_control"] if negative["incorrectly_passes_project_sync_gate"] is not False: return False if abs(negative["offset_seconds"]) <= 0.08 and negative["confidence"] > 3: return False import re def words(value): expanded = re.sub(r"\bi['’]m\b", "i am", value.lower()) return re.sub(r"[^a-z0-9 ]", " ", expanded).split() expected = words(asr["expected_text"]) return bool(expected) and expected == words(asr["master_transcript"]) == words(asr["returned_transcript"]) except (KeyError, TypeError, ValueError, AttributeError): return False def plan(request, config, ledger): errors = [] stage = request.get("stage") if stage not in {"still", "motion_test", "final", "upscale_or_edit"}: errors.append("Unknown production stage.") if not request.get("shot_id"): errors.append("A stable shot_id is required across retries.") brief = request.get("brief", {}) for field in ("scene", "wardrobe", "action", "camera", "lighting", "aspect_ratio", "duration_seconds"): if not brief.get(field): errors.append(f"Shot brief missing {field}.") if not request.get("prompt"): errors.append("Exact prompt is required.") references = request.get("reference_sha256", []) if not references or any(not SHA256.fullmatch(value or "") for value in references): errors.append("Hash actual reference bytes before planning.") renderer = request.get("renderer", {}) required = ("provider", "model_id", "model_version", "resolution", "minimum_duration_seconds", "audio_features", "settings", "rate_unit", "source_url", "verified_at") for field in required: if renderer.get(field) is None or renderer.get(field) == "": errors.append(f"Renderer metadata missing {field}.") if renderer.get("access_verified") is not True: errors.append("Provider/model access is not verified; no paid request is allowed.") if not isinstance(renderer.get("rate_usd_micros"), int) or isinstance(renderer.get("rate_usd_micros"), bool) or renderer.get("rate_usd_micros", -1) < 0: errors.append("Verified rate in integer USD micros is required.") duration = brief.get("duration_seconds") minimum = renderer.get("minimum_duration_seconds") if isinstance(duration, (int, float)) and isinstance(minimum, (int, float)) and duration < minimum: errors.append("Duration is below the provider's verified minimum.") key = cache_key(request) matches = [e for e in ledger.get("entries", []) if e.get("cache_key") == key] cached = next((e for e in matches if e.get("status") == "accepted" and e.get("qa_passed") is True and SHA256.fullmatch(e.get("output_sha256", ""))), None) if cached: return {"allowed": False, "action": "reuse_cache", "cache_key": key, "entry_id": cached["id"], "errors": []} existing = next((e for e in matches if e.get("status") in {"reserved", "submitted", "rendered", "accepted"}), None) if existing: return {"allowed": False, "action": "review_existing", "cache_key": key, "entry_id": existing["id"], "errors": []} if any(e.get("shot_id") == request.get("shot_id") and e.get("status") in {"reserved", "submitted"} for e in ledger.get("entries", [])): errors.append("A submission for this shot is unresolved; reconcile its receipt before any replacement request.") caps = config["caps"] if stage != "still": frame = request.get("first_frame", {}) if frame.get("review_status") != "accepted" or not frame.get("reviewer") or not SHA256.fullmatch(frame.get("sha256", "")): errors.append("Review and accept the exact scene first frame before motion.") if stage == "motion_test": attempts = [e for e in ledger.get("entries", []) if e.get("shot_id") == request.get("shot_id") and e.get("stage") == stage] if attempts and (request.get("retry_reviewed") is not True or not request.get("retry_reviewer")): errors.append("A retry needs recorded review; no automatic paid re-POST.") if len(attempts) >= caps["max_motion_test_attempts_per_shot"]: errors.append("Two motion-test attempts used; stop for review.") if renderer.get("resolution") != caps["motion_test_resolution"]: errors.append(f"Use the configured short {caps['motion_test_resolution']} motion test.") if isinstance(duration, (int, float)) and duration > caps.get("motion_test_max_seconds", 5): errors.append("Test duration exceeds the configured short-test limit.") if stage == "final": accepted_test = next((e for e in ledger.get("entries", []) if e.get("cache_key") == request.get("accepted_test_cache_key") and e.get("stage") == "motion_test" and e.get("status") == "accepted" and e.get("shot_id") == request.get("shot_id") and e.get("first_frame_sha256") == request.get("first_frame", {}).get("sha256") and e.get("qa_passed") is True and e.get("shot_signature") == shot_signature(request) and SHA256.fullmatch(e.get("output_sha256", ""))), None) if not accepted_test: errors.append("Final render needs an accepted matching motion test with passed QA.") if request.get("higher_resolution_needed") is not True: errors.append("Higher-resolution final is not needed; keep the accepted test.") if renderer.get("resolution") != caps["final_resolution"]: errors.append("Use the configured 1080p final; no mass 4K rerolls.") if stage == "upscale_or_edit": finished = [e for e in ledger.get("entries", []) if e.get("shot_id") == request.get("shot_id") and e.get("stage") == stage] if len(finished) >= caps["max_upscale_or_edit_passes"]: errors.append("One finishing pass used; stop for review.") accepted_input = next((e for e in ledger.get("entries", []) if e.get("cache_key") == request.get("accepted_input_cache_key") and e.get("shot_id") == request.get("shot_id") and e.get("status") == "accepted" and e.get("qa_passed") is True and SHA256.fullmatch(e.get("output_sha256", ""))), None) if not request.get("finishing_reason") or request.get("accepted_input") is not True or not accepted_input: errors.append("Finishing needs a matching accepted output and a specific need.") quote = request.get("estimated_external_cost_usd_micros") rate = renderer.get("rate_usd_micros") unit = renderer.get("rate_unit") if isinstance(rate, int) and not isinstance(rate, bool) and isinstance(quote, int): if unit in {"request", "clip"}: if quote < rate: errors.append("Reserved quote is below the verified per-request rate.") elif unit in {"second", "per_second"} and isinstance(duration, (int, float)): from decimal import Decimal, ROUND_CEILING base_cost = int((Decimal(rate) * Decimal(str(duration))).to_integral_value(rounding=ROUND_CEILING)) if quote < base_cost: errors.append("Reserved quote is below the verified duration rate.") else: errors.append("Normalize the verified price to request or second units before reserving.") if not isinstance(quote, int) or isinstance(quote, bool) or quote < 0: errors.append("Reserve a verified API cost upper bound before spending.") else: committed = summarize(ledger)["external_committed_usd_micros"] if committed + quote > caps["external_api_total_usd_micros"]: errors.append("Shared $5 external API cap would be exceeded; no top-up.") return {"allowed": not errors, "action": "reserve" if not errors else "stop_for_review", "cache_key": key, "errors": errors} def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("command", choices=("cache-key", "plan", "reserve", "summarize")) parser.add_argument("request", nargs="?") parser.add_argument("--config", default="production.json") parser.add_argument("--ledger", default="production-ledger.json") args = parser.parse_args() ledger_path = Path(args.ledger) if args.command == "summarize": ledger = json.loads(ledger_path.read_text()) print(json.dumps(summarize(ledger), indent=2)) return request = json.loads(Path(args.request).read_text()) if args.command == "cache-key": print(cache_key(request)) return config = json.loads(Path(args.config).read_text()) if args.command == "reserve": import fcntl with ledger_path.open("r+", encoding="utf-8") as handle: fcntl.flock(handle, fcntl.LOCK_EX) ledger = json.load(handle) result = plan(request, config, ledger) if result["allowed"]: ledger["entries"].append({ "id": result["cache_key"], "shot_id": request["shot_id"], "stage": request["stage"], "cache_key": result["cache_key"], "shot_signature": shot_signature(request), "status": "reserved", "qa_passed": False, "first_frame_sha256": request.get("first_frame", {}).get("sha256"), "estimated_external_cost_usd_micros": request["estimated_external_cost_usd_micros"], "cost_finalized": False, "cost_components": [], "request": request, "provider_job_id": None, "output_sha256": None, }) ledger["summary"] = summarize(ledger) handle.seek(0) handle.write(json.dumps(ledger, ensure_ascii=False, indent=2) + "\n") handle.truncate() handle.flush() else: ledger = json.loads(ledger_path.read_text()) result = plan(request, config, ledger) print(json.dumps(result, indent=2)) if not result["allowed"]: raise SystemExit(2) if __name__ == "__main__": main()