Uncategorized

The balance of power in the digital age is shifting. While governments and large corporations have long used data to track individuals, a new open-source project called OpenPlanter is giving that power back to the public. Created by a developer ‘Shin Megami Boson‘, OpenPlanter is a recursive-language-model investigation agent. Its goal is simple: help you keep tabs on your government, since they are almost certainly keeping tabs on you. Solving the ‘Heterogeneous Data’ Problem Investigative work is difficult because data is messy. Public records are often spread across 100 different formats. You might have a CSV of campaign finance records, a JSON file of government contracts, and a PDF of lobbying disclosures. OpenPlanter ingests these disparate structured and unstructured data sources effortlessly. It uses Large Language Models (LLMs) to perform entity resolution. This is the process of identifying when different records refer to the same person or company. Once it connects these dots, the agent probabilistically looks for anomalies. It searches for patterns that a human might miss, such as a sudden spike in contract wins following a specific lobbying event. The Architecture: Recursive Sub-Agent Delegation What makes OpenPlanter unique is its recursive engine. Most AI agents handle 1 request at a time. OpenPlanter, however, breaks large objectives into smaller pieces. If you give it a massive task, it uses a sub-agent delegation strategy. The agent has a default max-depth of 4. This means the main agent can spawn a sub-agent, which can spawn another, and so on. These agents work in parallel to: Resolve entities across massive datasets. Link datasets that have no common ID numbers. Construct evidence chains that back up every single finding. This recursive approach allows the system to handle investigations that are too large for a single ‘context window.’ The 2026 AI Stack OpenPlanter is built for the high-performance requirements of 2026. It is written in Python 3.10+ and integrates with the most advanced models available today. The technical documentation lists several supported providers: OpenAI: It uses gpt-5.2 as the default. Anthropic: It supports claude-opus-4-6. OpenRouter: It defaults to anthropic/claude-sonnet-4-5. Cerebras: It uses qwen-3-235b-a22b-instruct-2507 for high-speed tasks. The system also uses Exa for web searches and Voyage for high-accuracy embeddings. This multi-model strategy ensures that the agent uses the best ‘brain’ for each specific sub-task. 19 Tools for Digital Forensics The agent is equipped with 19 specialized tools. These tools allow it to interact with the real world rather than just ‘chatting.’ These are organized into 4 core areas: File I/O and Workspace: Tools like read_file, write_file, and hashline_edit allow the agent to manage its own database of findings. Shell Execution: The agent can use run_shell to execute actual code. It can write a Python script to analyze a dataset and then run that script to get results. Web Retrieval: With web_search and fetch_url, it can pull live data from government registries or news sites. Planning and Logic: The think tool lets the agent pause and strategize. It uses acceptance-criteria to verify that a sub-task was completed correctly before moving to the next step. Deployment and Interface OpenPlanter is designed to be accessible but powerful. It features a Terminal User Interface (TUI) built with rich and prompt_toolkit. The interface includes a splash art screen of ASCII potted plants, but the work it does is serious. You can get started quickly using Docker. By running docker compose up, the agent starts in a container. This is a critical security feature because it isolates the agent’s run_shell commands from the user’s host operating system. The command-line interface allows for ‘headless’ tasks. You can run a single command like: Copy CodeCopiedUse a different Browser openplanter-agent –task “Flag all vendor overlaps in lobbying data” –workspace ./data The agent will then work autonomously until it produces a final report. Key Takeaways Autonomous Recursive Logic: Unlike standard agents, OpenPlanter uses a recursive sub-agent delegation strategy (default max-depth of 4). It breaks complex investigative objectives into smaller sub-tasks, parallelizing work across multiple agents to build detailed evidence chains. Heterogeneous Data Correlation: The agent is built to ingest and resolve disparate structured and unstructured data. It can simultaneously process CSV files, JSON records, and unstructured text (like PDFs) to identify entities across fragmented datasets. Probabilistic Anomaly Detection: By performing entity resolution, OpenPlanter automatically connects records—such as matching a corporate alias to a lobbying disclosure—and looks for probabilistic anomalies to surface hidden connections between government spending and private interests. High-End 2026 Model Stack: The system is provider-agnostic and utilizes the latest frontier models, including OpenAI gpt-5.2, Anthropic claude-opus-4-6, and Cerebras qwen-3-235b-a22b-instruct-2507 for high-speed inference. Integrated Toolset for Forensics: OpenPlanter features 19 distinct tools, including shell execution (run_shell), web search (Exa), and file patching (hashline_edit). This allows it to write and run its own analysis scripts while verifying results against real-world acceptance criteria. Check out the Repo here. Also, feel free to follow us on Twitter and don’t forget to join our 100k+ ML SubReddit and Subscribe to our Newsletter. Wait! are you on telegram? now you can join us on telegram as well. Disclaimer: MarkTechPost does not endorse the OpenPlanter project and provides this technical report for informational purposes only. The post Is There a Community Edition of Palantir? Meet OpenPlanter: An Open Source Recursive AI Agent for Your Micro Surveillance Use Cases appeared first on MarkTechPost.

2025 was a year of reckoning, including how the heads of the top AI companies made promises they couldn’t keep. In this exclusive subscriber-only eBook, you’ll learn more about why we may need to readjust our expectations. This story is part of the Hype Correction package.by Will Douglas Heaven December 15, 2025 ACCESS EBOOK Table of Contents: LLMs are not everything AI is not a quick fix to all your problems Are we in a bubble? (If so, what kind of bubble?) ChatGPT was not the beginning, and it won’t be the end Related Stories: The great AI hype correction of 2025 An MIT Technology Review series: Hype Correction Access all subscriber-only eBooks: ACCESS ALL EBOOKS

In this tutorial, we design a practical image-generation workflow using the Diffusers library. We start by stabilizing the environment, then generate high-quality images from text prompts using Stable Diffusion with an optimized scheduler. We accelerate inference with a LoRA-based latent consistency approach, guide composition with ControlNet under edge conditioning, and finally perform localized edits via inpainting. Also, we focus on real-world techniques that balance image quality, speed, and controllability. Copy CodeCopiedUse a different Browser !pip -q uninstall -y pillow Pillow || true !pip -q install –upgrade –force-reinstall “pillow<12.0” !pip -q install –upgrade diffusers transformers accelerate safetensors huggingface_hub opencv-python import os, math, random import torch import numpy as np import cv2 from PIL import Image, ImageDraw, ImageFilter from diffusers import ( StableDiffusionPipeline, StableDiffusionInpaintPipeline, ControlNetModel, StableDiffusionControlNetPipeline, UniPCMultistepScheduler, ) We prepare a clean and compatible runtime by resolving dependency conflicts and installing all required libraries. We ensure image processing works reliably by pinning the correct Pillow version and loading the Diffusers ecosystem. We also import all core modules needed for generation, control, and inpainting workflows. Copy CodeCopiedUse a different Browser def seed_everything(seed=42): random.seed(seed) np.random.seed(seed) torch.manual_seed(seed) torch.cuda.manual_seed_all(seed) def to_grid(images, cols=2, bg=255): if isinstance(images, Image.Image): images = [images] w, h = images[0].size rows = math.ceil(len(images) / cols) grid = Image.new(“RGB”, (cols*w, rows*h), (bg, bg, bg)) for i, im in enumerate(images): grid.paste(im, ((i % cols)*w, (i // cols)*h)) return grid device = “cuda” if torch.cuda.is_available() else “cpu” dtype = torch.float16 if device == “cuda” else torch.float32 print(“device:”, device, “| dtype:”, dtype) We define utility functions to ensure reproducibility and to organize visual outputs efficiently. We set global random seeds so our generations remain consistent across runs. We also detect the available hardware and configure precision to optimize performance on the GPU or CPU. Copy CodeCopiedUse a different Browser seed_everything(7) BASE_MODEL = “runwayml/stable-diffusion-v1-5” pipe = StableDiffusionPipeline.from_pretrained( BASE_MODEL, torch_dtype=dtype, safety_checker=None, ).to(device) pipe.scheduler = UniPCMultistepScheduler.from_config(pipe.scheduler.config) if device == “cuda”: pipe.enable_attention_slicing() pipe.enable_vae_slicing() prompt = “a cinematic photo of a futuristic street market at dusk, ultra-detailed, 35mm, volumetric lighting” negative_prompt = “blurry, low quality, deformed, watermark, text” img_text = pipe( prompt=prompt, negative_prompt=negative_prompt, num_inference_steps=25, guidance_scale=6.5, width=768, height=512, ).images[0] We initialize the base Stable Diffusion pipeline and switch to a more efficient UniPC scheduler. We generate a high-quality image directly from a text prompt using carefully chosen guidance and resolution settings. This establishes a strong baseline for subsequent improvements in speed and control. Copy CodeCopiedUse a different Browser LCM_LORA = “latent-consistency/lcm-lora-sdv1-5” pipe.load_lora_weights(LCM_LORA) try: pipe.fuse_lora() lora_fused = True except Exception as e: lora_fused = False print(“LoRA fuse skipped:”, e) fast_prompt = “a clean product photo of a minimal smartwatch on a reflective surface, studio lighting” fast_images = [] for steps in [4, 6, 8]: fast_images.append( pipe( prompt=fast_prompt, negative_prompt=negative_prompt, num_inference_steps=steps, guidance_scale=1.5, width=768, height=512, ).images[0] ) grid_fast = to_grid(fast_images, cols=3) print(“LoRA fused:”, lora_fused) W, H = 768, 512 layout = Image.new(“RGB”, (W, H), “white”) draw = ImageDraw.Draw(layout) draw.rectangle([40, 80, 340, 460], outline=”black”, width=6) draw.ellipse([430, 110, 720, 400], outline=”black”, width=6) draw.line([0, 420, W, 420], fill=”black”, width=5) edges = cv2.Canny(np.array(layout), 80, 160) edges = np.stack([edges]*3, axis=-1) canny_image = Image.fromarray(edges) CONTROLNET = “lllyasviel/sd-controlnet-canny” controlnet = ControlNetModel.from_pretrained( CONTROLNET, torch_dtype=dtype, ).to(device) cn_pipe = StableDiffusionControlNetPipeline.from_pretrained( BASE_MODEL, controlnet=controlnet, torch_dtype=dtype, safety_checker=None, ).to(device) cn_pipe.scheduler = UniPCMultistepScheduler.from_config(cn_pipe.scheduler.config) if device == “cuda”: cn_pipe.enable_attention_slicing() cn_pipe.enable_vae_slicing() cn_prompt = “a modern cafe interior, architectural render, soft daylight, high detail” img_controlnet = cn_pipe( prompt=cn_prompt, negative_prompt=negative_prompt, image=canny_image, num_inference_steps=25, guidance_scale=6.5, controlnet_conditioning_scale=1.0, ).images[0] We accelerate inference by loading and fusing a LoRA adapter and demonstrate fast sampling with very few diffusion steps. We then construct a structural conditioning image and apply ControlNet to guide the layout of the generated scene. This allows us to preserve composition while still benefiting from creative text guidance. Copy CodeCopiedUse a different Browser mask = Image.new(“L”, img_controlnet.size, 0) mask_draw = ImageDraw.Draw(mask) mask_draw.rectangle([60, 90, 320, 170], fill=255) mask = mask.filter(ImageFilter.GaussianBlur(2)) inpaint_pipe = StableDiffusionInpaintPipeline.from_pretrained( BASE_MODEL, torch_dtype=dtype, safety_checker=None, ).to(device) inpaint_pipe.scheduler = UniPCMultistepScheduler.from_config(inpaint_pipe.scheduler.config) if device == “cuda”: inpaint_pipe.enable_attention_slicing() inpaint_pipe.enable_vae_slicing() inpaint_prompt = “a glowing neon sign that says ‘CAFÉ’, cyberpunk style, realistic lighting” img_inpaint = inpaint_pipe( prompt=inpaint_prompt, negative_prompt=negative_prompt, image=img_controlnet, mask_image=mask, num_inference_steps=30, guidance_scale=7.0, ).images[0] os.makedirs(“outputs”, exist_ok=True) img_text.save(“outputs/text2img.png”) grid_fast.save(“outputs/lora_fast_grid.png”) layout.save(“outputs/layout.png”) canny_image.save(“outputs/canny.png”) img_controlnet.save(“outputs/controlnet.png”) mask.save(“outputs/mask.png”) img_inpaint.save(“outputs/inpaint.png”) print(“Saved outputs:”, sorted(os.listdir(“outputs”))) print(“Done.”) We create a mask to isolate a specific region and apply inpainting to modify only that part of the image. We refine the selected area using a targeted prompt while keeping the rest intact. Finally, we save all intermediate and final outputs to disk for inspection and reuse. In conclusion, we demonstrated how a single Diffusers pipeline can evolve into a flexible, production-ready image generation system. We explained how to move from pure text-to-image generation to fast sampling, structural control, and targeted image editing without changing frameworks or tooling. This tutorial highlights how we can combine schedulers, LoRA adapters, ControlNet, and inpainting to create controllable and efficient generative pipelines that are easy to extend for more advanced creative or applied use cases. Check out the Full Codes here. Also, feel free to follow us on Twitter and don’t forget to join our 100k+ ML SubReddit and Subscribe to our Newsletter. Wait! are you on telegram? now you can join us on telegram as well. The post A Coding Guide to High-Quality Image Generation, Control, and Editing Using HuggingFace Diffusers appeared first on MarkTechPost.

There’s a measles outbreak happening close to where I live. Since the start of this year, 34 cases have been confirmed in Enfield, a northern borough of London. Most of those affected are children under the age of 11. One in five have needed hospital treatment. It’s another worrying development for an incredibly contagious and potentially fatal disease. Since October last year, 962 cases of measles have been confirmed in South Carolina. Large outbreaks (with more than 50 confirmed cases) are underway in four US states. Smaller outbreaks are being reported in another 12 states. The vast majority of these cases have been children who were not fully vaccinated. Vaccine hesitancy is thought to be a significant reason children are missing out on important vaccines—the World Health Organization described it as one of the 10 leading threats to global health in 2019. And if we’re seeing more measles cases now, we might expect to soon see more cases of other vaccine-preventable infections, including some that can cause liver cancer or meningitis. Some people will always argue that measles is not a big deal—that infections used to be common, and most people survived them and did just fine. It is true that in most cases kids do recover well from the virus. But not always. Measles symptoms tend to start with a fever and a runny nose. The telltale rash comes later. In some cases, severe complications develop. They can include pneumonia, blindness, and inflammation of the brain. Some people won’t develop complications until years later. In rare cases, the disease can be fatal. Before the measles vaccine was introduced, in 1963, measles epidemics occurred every two to three years, according to the WHO. Back then, around 2.6 million people died from measles every year. Since it was introduced, the measles vaccine is thought to have prevented almost 59 million deaths. But vaccination rates have been lagging, says Anne Zink, an emergency medicine physician and clinical fellow at the Yale School of Public Health. “We’ve seen a slow decline in people who are willing to get vaccinated against measles for some time,” she says. “As we get more and more people who are at risk because they’re unvaccinated, the higher the chances that the disease can then spread and take off.” Vaccination rates need to be at 95% to prevent measles outbreaks. But rates are well below that level in some regions. Across South Carolina, the proportion of kindergartners who received both doses of the MMR vaccine, which protects against measles as well as mumps and rubella, has dropped steadily over the last five years, from 94% in 2020-2021 to 91% in 2024-2025. Some schools in the state have coverage rates as low as 20%, state epidemiologist Linda Bell told reporters last month. Vaccination rates are low in London, too. Fewer than 70% of children have received both doses of their MMR by the time they turn five, according to the UK Health Security Agency. In some boroughs, vaccination rates are as low as 58%. So perhaps it’s not surprising we’re seeing outbreaks. The UK is one of six countries to have lost their measles elimination status last month, along with Spain, Austria, Armenia, Azerbaijan, and Uzbekistan. Canada lost its elimination status last year. The highly contagious measles could be a bellwether for other vaccine-preventable diseases. Zink is already seeing signs. She points to a case of polio that paralyzed a man in New York in 2022. That happened when rates of polio vaccination were low, she says. “Polio is a great example of … a disease that is primarily asymptomatic, and most people don’t have any symptoms whatsoever, but for the people who do get symptoms, it can be life-threatening.” Then there’s mumps—another disease the MMR vaccine protects against. It’s another one of those infections that can be symptom-free and harmless in some, especially children, but nasty for others. It can cause a painful swelling of the testes, and other complications include brain swelling and deafness. (From my personal experience of being hospitalized with mumps, I can attest that even “mild” infections are pretty horrible.) Mumps is less contagious than measles, so we might expect a delay between an uptick in measles cases and the spread of mumps, says Zink. But she says that she’s more concerned about hepatitis B. “It lives on surfaces for a long period of time, and if you’re not vaccinated against it and you’re exposed to it as a kid, you’re at a really high risk of developing liver cancer and death,” she says. Zink was formerly chief medical officer of Alaska, a state that in the 1970s had the world’s highest rate of childhood liver cancer caused by hepatitis B. Screening and universal newborn vaccination programs eliminated the virus’s spread. Public health experts worry that the current US administration’s position on vaccines may contribute to the decline in vaccine uptake. Last month the US Centers for Disease Control and Prevention approved changes to childhood vaccination recommendations. The agency no longer recommends the hepatitis B vaccine for all newborns. The chair of the CDC’s vaccine advisory panel has also questioned broad vaccine recommendations for polio. Even vitamin injections are being refused by parents, says Zink. A shot of vitamin K at birth can help prevent severe bleeding in some babies. But recent research suggests that parents of 5% of newborns are refusing it (up from 2.9% in 2017). “I can’t tell you how many of my pediatric [doctor] friends have told me about having to care for a kiddo in the ICU with … bleeding into their brain because the kid didn’t get vitamin K at birth,” says Zink. “And that can kill kids, [or have] lifelong, devastating, stroke-like symptoms.” All this paints a pretty bleak picture for children’s health. But things can change. Vaccination can still offer protection to plenty of people at risk of infection. South Carolina’s Department of Public Health is offering free MMR vaccinations to residents at mobile clinics. “It’s easy to think ‘It’s not going to be me,’” says Zink. “Seeing kiddos

NVIDIA has just released Dynamo v0.9.0. This is the most significant infrastructure upgrade for the distributed inference framework to date. This update simplifies how large-scale models are deployed and managed. The release focuses on removing heavy dependencies and improving how GPUs handle multi-modal data. The Great Simplification: Removing NATS and etcd The biggest change in v0.9.0 is the removal of NATS and ETCD. In previous versions, these tools handled service discovery and messaging. However, they added ‘operational tax’ by requiring developers to manage extra clusters. NVIDIA replaced these with a new Event Plane and a Discovery Plane. The system now uses ZMQ (ZeroMQ) for high-performance transport and MessagePack for data serialization. For teams using Kubernetes, Dynamo now supports Kubernetes-native service discovery. This change makes the infrastructure leaner and easier to maintain in production environments. Multi-Modal Support and the E/P/D Split Dynamo v0.9.0 expands multi-modal support across 3 main backends: vLLM, SGLang, and TensorRT-LLM. This allows models to process text, images, and video more efficiently. A key feature in this update is the E/P/D (Encode/Prefill/Decode) split. In standard setups, a single GPU often handles all 3 stages. This can cause bottlenecks during heavy video or image processing. v0.9.0 introduces Encoder Disaggregation. You can now run the Encoder on a separate set of GPUs from the Prefill and Decode workers. This allows you to scale your hardware based on the specific needs of your model. Sneak Preview: FlashIndexer This release includes a sneak preview of FlashIndexer. This component is designed to solve latency issues in distributed KV cache management. When working with large context windows, moving Key-Value (KV) data between GPUs is a slow process. FlashIndexer improves how the system indexes and retrieves these cached tokens. This results in a lower Time to First Token (TTFT). While still a preview, it represents a major step toward making distributed inference feel as fast as local inference. Smart Routing and Load Estimation Managing traffic across 100s of GPUs is difficult. Dynamo v0.9.0 introduces a smarter Planner that uses predictive load estimation. The system uses a Kalman filter to predict the future load of a request based on past performance. It also supports routing hints from the Kubernetes Gateway API Inference Extension (GAIE). This allows the network layer to communicate directly with the inference engine. If a specific GPU group is overloaded, the system can route new requests to idle workers with higher precision. The Technical Stack at a Glance The v0.9.0 release updates several core components to their latest stable versions. Here is the breakdown of the supported backends and libraries: Component Version vLLM v0.14.1 SGLang v0.5.8 TensorRT-LLM v1.3.0rc1 NIXL v0.9.0 Rust Core dynamo-tokens crate The inclusion of the dynamo-tokens crate, written in Rust, ensures that token handling remains high-speed. For data transfer between GPUs, Dynamo continues to leverage NIXL (NVIDIA Inference Transfer Library) for RDMA-based communication. Key Takeaways Infrastructure Decoupling (Goodbye NATS and ETCD): The release completes the modernization of the communication architecture. By replacing NATS and ETCD with a new Event Plane (using ZMQ and MessagePack) and Kubernetes-native service discovery, the system removes the ‘operational tax’ of managing external clusters. Full Multi-Modal Disaggregation (E/P/D Split): Dynamo now supports a complete Encode/Prefill/Decode (E/P/D) split across all 3 backends (vLLM, SGLang, and TRT-LLM). This allows you to run vision or video encoders on separate GPUs, preventing compute-heavy encoding tasks from bottlenecking the text generation process. FlashIndexer Preview for Lower Latency :The ‘sneak preview’ of FlashIndexer introduces a specialized component to optimize distributed KV cache management. It is designed to make the indexing and retrieval of conversation ‘memory’ significantly faster, aimed at further reducing the Time to First Token (TTFT). Smarter Scheduling with Kalman Filters: The system now uses predictive load estimation powered by Kalman filters. This allows the Planner to forecast GPU load more accurately and handle traffic spikes proactively, supported by routing hints from the Kubernetes Gateway API Inference Extension (GAIE). Check out the GitHub Release here. Also, feel free to follow us on Twitter and don’t forget to join our 100k+ ML SubReddit and Subscribe to our Newsletter. Wait! are you on telegram? now you can join us on telegram as well. The post NVIDIA Releases Dynamo v0.9.0: A Massive Infrastructure Overhaul Featuring FlashIndexer, Multi-Modal Support, and Removed NATS and ETCD appeared first on MarkTechPost.

In this tutorial, we build a glass-box agentic workflow that makes every decision traceable, auditable, and explicitly governed by human approval. We design the system to log each thought, action, and observation into a tamper-evident audit ledger while enforcing dynamic permissioning for high-risk operations. By combining LangGraph’s interrupt-driven human-in-the-loop control with a hash-chained database, we demonstrate how agentic systems can move beyond opaque automation and align with modern governance expectations. Throughout the tutorial, we focus on practical, runnable patterns that turn governance from an afterthought into a first-class system feature. Copy CodeCopiedUse a different Browser !pip -q install -U langgraph langchain-core openai “pydantic<=2.12.3” import os import json import time import hmac import hashlib import secrets import sqlite3 import getpass from typing import Any, Dict, List, Optional, Literal, TypedDict from openai import OpenAI from langchain_core.messages import SystemMessage, HumanMessage, AIMessage from langgraph.graph import StateGraph, END from langgraph.types import interrupt, Command if not os.getenv(“OPENAI_API_KEY”): os.environ[“OPENAI_API_KEY”] = getpass.getpass(“Enter OpenAI API Key: “) client = OpenAI() MODEL = “gpt-5” We install all required libraries and import the core modules needed for agentic workflows and governance. We securely collect the OpenAI API key through a terminal prompt to avoid hard-coding secrets in the notebook. We also initialize the OpenAI client and define the model that drives the agent’s reasoning loop. Copy CodeCopiedUse a different Browser CREATE_SQL = “”” CREATE TABLE IF NOT EXISTS audit_log ( id INTEGER PRIMARY KEY AUTOINCREMENT, ts_unix INTEGER NOT NULL, actor TEXT NOT NULL, event_type TEXT NOT NULL, payload_json TEXT NOT NULL, prev_hash TEXT NOT NULL, row_hash TEXT NOT NULL ); CREATE TABLE IF NOT EXISTS ot_tokens ( token_id TEXT PRIMARY KEY, token_hash TEXT NOT NULL, purpose TEXT NOT NULL, expires_unix INTEGER NOT NULL, used INTEGER NOT NULL DEFAULT 0 ); “”” def _sha256_hex(s: bytes) -> str: return hashlib.sha256(s).hexdigest() def _canonical_json(obj: Any) -> str: return json.dumps(obj, sort_keys=True, separators=(“,”, “:”), ensure_ascii=False) class AuditLedger: def __init__(self, path: str = “glassbox_audit.db”): self.conn = sqlite3.connect(path, check_same_thread=False) self.conn.executescript(CREATE_SQL) self.conn.commit() def _last_hash(self) -> str: row = self.conn.execute(“SELECT row_hash FROM audit_log ORDER BY id DESC LIMIT 1”).fetchone() return row[0] if row else “GENESIS” def append(self, actor: str, event_type: str, payload: Any) -> int: ts = int(time.time()) prev_hash = self._last_hash() payload_json = _canonical_json(payload) material = f”{ts}|{actor}|{event_type}|{payload_json}|{prev_hash}”.encode(“utf-8”) row_hash = _sha256_hex(material) cur = self.conn.execute( “INSERT INTO audit_log (ts_unix, actor, event_type, payload_json, prev_hash, row_hash) VALUES (?, ?, ?, ?, ?, ?)”, (ts, actor, event_type, payload_json, prev_hash, row_hash), ) self.conn.commit() return cur.lastrowid def fetch_recent(self, limit: int = 50) -> List[Dict[str, Any]]: rows = self.conn.execute( “SELECT id, ts_unix, actor, event_type, payload_json, prev_hash, row_hash FROM audit_log ORDER BY id DESC LIMIT ?”, (limit,), ).fetchall() out = [] for r in rows[::-1]: out.append({ “id”: r[0], “ts_unix”: r[1], “actor”: r[2], “event_type”: r[3], “payload”: json.loads(r[4]), “prev_hash”: r[5], “row_hash”: r[6], }) return out def verify_integrity(self) -> Dict[str, Any]: rows = self.conn.execute( “SELECT id, ts_unix, actor, event_type, payload_json, prev_hash, row_hash FROM audit_log ORDER BY id ASC” ).fetchall() if not rows: return {“ok”: True, “rows”: 0, “message”: “Empty ledger.”} expected_prev = “GENESIS” for (id_, ts, actor, event_type, payload_json, prev_hash, row_hash) in rows: if prev_hash != expected_prev: return {“ok”: False, “at_id”: id_, “reason”: “prev_hash mismatch”} material = f”{ts}|{actor}|{event_type}|{payload_json}|{prev_hash}”.encode(“utf-8”) expected_hash = _sha256_hex(material) if not hmac.compare_digest(expected_hash, row_hash): return {“ok”: False, “at_id”: id_, “reason”: “row_hash mismatch”} expected_prev = row_hash return {“ok”: True, “rows”: len(rows), “message”: “Hash chain valid.”} ledger = AuditLedger() We design a hash-chained SQLite ledger that records every agent and system event in an append-only manner. We ensure each log entry cryptographically links to the previous one, making post-hoc tampering detectable. We also provide utilities to inspect recent events and verify the integrity of the entire audit chain. Copy CodeCopiedUse a different Browser def mint_one_time_token(purpose: str, ttl_seconds: int = 600) -> Dict[str, str]: token_id = secrets.token_hex(12) token_plain = secrets.token_urlsafe(20) token_hash = _sha256_hex(token_plain.encode(“utf-8”)) expires = int(time.time()) + ttl_seconds ledger.conn.execute( “INSERT INTO ot_tokens (token_id, token_hash, purpose, expires_unix, used) VALUES (?, ?, ?, ?, 0)”, (token_id, token_hash, purpose, expires), ) ledger.conn.commit() return {“token_id”: token_id, “token_plain”: token_plain, “purpose”: purpose, “expires_unix”: str(expires)} def consume_one_time_token(token_id: str, token_plain: str, purpose: str) -> bool: row = ledger.conn.execute( “SELECT token_hash, purpose, expires_unix, used FROM ot_tokens WHERE token_id = ?”, (token_id,), ).fetchone() if not row: return False token_hash_db, purpose_db, expires_unix, used = row if used == 1: return False if purpose_db != purpose: return False if int(time.time()) > int(expires_unix): return False token_hash_in = _sha256_hex(token_plain.encode(“utf-8”)) if not hmac.compare_digest(token_hash_in, token_hash_db): return False ledger.conn.execute(“UPDATE ot_tokens SET used = 1 WHERE token_id = ?”, (token_id,)) ledger.conn.commit() return True def tool_financial_transfer(amount_usd: float, to_account: str) -> Dict[str, Any]: return {“status”: “success”, “transfer_id”: “tx_” + secrets.token_hex(6), “amount_usd”: amount_usd, “to_account”: to_account} def tool_rig_move(rig_id: str, direction: Literal[“UP”, “DOWN”], meters: float) -> Dict[str, Any]: return {“status”: “success”, “rig_event_id”: “rig_” + secrets.token_hex(6), “rig_id”: rig_id, “direction”: direction, “meters”: meters} We implement a secure, single-use token mechanism that enables human approval for high-risk actions. We generate time-limited tokens, store only their hashes, and invalidate them immediately after use. We also define simulated restricted tools that represent sensitive operations such as financial transfers or physical rig movements. Copy CodeCopiedUse a different Browser RestrictedTool = Literal[“financial_transfer”, “rig_move”, “none”] class GlassBoxState(TypedDict): messages: List[Any] proposed_tool: RestrictedTool tool_args: Dict[str, Any] last_observation: Optional[Dict[str, Any]] SYSTEM_POLICY = “””You are a governance-first agent. You MUST propose actions in a structured JSON format with these keys: – thought – action – args Return ONLY JSON.””” def llm_propose_action(messages: List[Any]) -> Dict[str, Any]: input_msgs = [{“role”: “system”, “content”: SYSTEM_POLICY}] for m in messages: if isinstance(m, SystemMessage): input_msgs.append({“role”: “system”, “content”: m.content}) elif isinstance(m, HumanMessage): input_msgs.append({“role”: “user”, “content”: m.content}) elif isinstance(m, AIMessage): input_msgs.append({“role”: “assistant”, “content”: m.content}) resp = client.responses.create(model=MODEL, input=input_msgs) txt = resp.output_text.strip() try: return json.loads(txt) except Exception: return {“thought”: “fallback”, “action”: “ask_human”, “args”: {}} def node_think(state: GlassBoxState) -> GlassBoxState: proposal = llm_propose_action(state[“messages”]) ledger.append(“agent”, “THOUGHT”, {“thought”: proposal.get(“thought”)}) ledger.append(“agent”, “ACTION”, proposal) action = proposal.get(“action”, “no_op”) args = proposal.get(“args”, {}) if action in [“financial_transfer”, “rig_move”]: state[“proposed_tool”] = action state[“tool_args”] = args else: state[“proposed_tool”] = “none” state[“tool_args”] = {} return state def node_permission_gate(state: GlassBoxState) -> GlassBoxState: if state[“proposed_tool”] == “none”: return state token = mint_one_time_token(state[“proposed_tool”]) payload = {“token_id”: token[“token_id”], “token_plain”: token[“token_plain”]} human_input = interrupt(payload) state[“tool_args”][“_token_id”] = token[“token_id”] state[“tool_args”][“_human_token_plain”] = str(human_input) return state