System Architecture
The “Tribunal” System
Interactive visualisation of the Parallax Protocol workflow
High-Level Architecture
The Orchestrator
Python/Go - Central brain managing the lifecycle
Generate→Compile→Execute→Vote→Patch
C
Agent A
C Smith
J
Agent B
Java Architect
Py
Agent C
Python Scribe
Consensus
Voting Engine
Proving Grounds (Docker Containers)
gcc
compile + run
javac + java
compile + run
python3
interpret
Observatory
ELK + Grafana
The Workflow: Step by Step
1
The Fabrication
Parallel Generation
The Orchestrator reads the Golden Spec and spins up three distinct LLM contexts (Agents), ensuring they do not share memory to prevent contamination of logic.
C
Agent A
The C Smith
Generates C code + Makefile
Java
Agent B
The Java Architect
Generates Java Class
Py
Agent C
The Python Scribe
Generates Python script
Reference Implementation: The Arbiter
arbiter.py
import subprocess
import concurrent.futures
from collections import Counter
# --- Configuration ---
PROGRAMS = {
"c": {"compile": "gcc -o calc calc.c", "run": "./calc"},
"java": {"compile": "javac Calc.java", "run": "java Calc"},
"python": {"compile": None, "run": "python3 calc.py"}
}
INPUT_DATA = "1000 5 2" # Principal, Rate, Time
def compile_and_run(lang, config, input_args):
"""Compiles (if needed) and runs the specific language implementation."""
# 1. Compile Phase
if config["compile"]:
compile_proc = subprocess.run(config["compile"], shell=True, capture_output=True)
if compile_proc.returncode != 0:
return lang, None, f"Compile Error: {compile_proc.stderr.decode()}"
# 2. Execution Phase
try:
cmd = f"{config['run']} {input_args}"
run_proc = subprocess.run(cmd, shell=True, capture_output=True, text=True, timeout=2)
if run_proc.returncode != 0:
return lang, None, f"Runtime Error: {run_proc.stderr}"
return lang, run_proc.stdout.strip(), None
except Exception as e:
return lang, None, str(e)
def main():
print(f"--- Initiating Tribunal Protocol on Input: [{INPUT_DATA}] ---")
results = {}
# Parallel Execution using ThreadPool
with concurrent.futures.ThreadPoolExecutor(max_workers=3) as executor:
futures = []
for lang, config in PROGRAMS.items():
futures.append(executor.submit(compile_and_run, lang, config, INPUT_DATA))
for future in concurrent.futures.as_completed(futures):
lang, output, error = future.result()
if error:
print(f"[!] {lang.capitalize()} Error: {error}")
else:
results[lang] = output
print(f"[+] {lang.capitalize()} Output: {output}")
# Voting Logic
if not results:
print("All implementations failed!")
return
vote_counts = Counter(results.values())
consensus_value, votes = vote_counts.most_common(1)[0]
print("\n--- Tribunal Judgement ---")
if votes == len(PROGRAMS):
print(f"SUCCESS: Unanimous Consensus reached on result: {consensus_value}")
elif votes >= 2:
print(f"WARNING: Majority Consensus reached on result: {consensus_value}")
for lang, output in results.items():
if output != consensus_value:
print(f"[!!!] ANOMALY DETECTED IN {lang.upper()}")
if __name__ == "__main__":
main()