Prism Pulse | The Python Agentic Exchange

Structural Sieve

Codes are filtered through layers of security structures

Isolated Computation

While agents are relaying the function (s) of thier codes, airlock security features remain intact.

AI Interpretation of Codes

Prism Pulse uses Vertex AI to give semantic meaning to machine language

Code Grading Functionality

Prism Plus assigns a grade to each code injected, thus adding value by excelling in pools of similar scripts

Prism Pulse is a high-integrity code exchange where develpoers/software engineers inject autonomous agents into a verified ecosystem. Every submission undergoes deep AST (Abstract Syntax Tree) analysis to neutralize malicious intent. Once vetted, logic is verified for performance in a secure sandbox, and graded.

Successful agents are vectored and indexed, becoming real-time upgrades for other autonomous entities. This is a breakthrough for the IT industry: Prism Pulse seamlessly syncs and refines code bases upon injection, all with one click, and developers earn money in the process.

How to Use (Step by Step Process Presented Below)

Notes on most recent upgrade which guards against Semantic Slippage:

-- Use descriptive argument names and ensure your function returns a predictable data type (like a dict, float, or list). Do not use variable-length arguments like *args or kwargs for core logic, as the AST scanner cannot strictly verify them.

--To guarantee that future upgrades to your agent behave exactly as you intended, you must submit a Python unittest file alongside your main logic script. This test suite acts as the "ground truth" for your logic.

⚠️ THE GOLDEN IMPORT RULE: You MUST import your function from solution.

When your code is injected into the Prism Pulse ecosystem, our secure Cloud Run Sandbox isolates your main logic script and temporarily renames it to solution.py.

Because of this, do not use your local filename in your test's import statement. For example, if your local file is named pharmacokinetic_clearance_model.py, the sandbox will crash with an ImportError if you try to import from that specific name.

Example test file:

import unittest

# ✅ CORRECT: The sandbox always recognizes your main script as 'solution'

from solution import my_custom_function

# ❌ INCORRECT: Do not use your local file name!

# from pharmacokinetic_clearance_model import my_custom_function

class TestMyLogic(unittest.TestCase):

def test_standard_case(self):

# Assert your expected inputs and outputs here

self.assertEqual(my_custom_function(10, 5), 50)

if __name__ == '__main__':

unittest.main()

Workflow after understanding why injecting test assertions are important and how to properly design test file for Prism Pulse:

1. Authenticate & Register

Login using the dashboard below. Once authenticated, you can purchase credit packages ($5, $10, or $20). Link your payment method via Stripe to claim USD for credits earned by your agents. Your dashboard displays your User ID, recent activity, and live credit balance.

2. Prepare & Inject ( must inject test assertions alongside) Use (inject.py)

Ensure your Python code is functional. Run the provided cmd (python3 inject.py 'your_script' --desc "") to submit your logic.

Security Sieve: Automatically scans for restricted modules.
Functional Sandbox: Validates runtime performance, and issues a grade based onthe following rubric:
## [A+] Elite / Production-Ready
* **Security:** 0 warnings; no unauthorized system calls.
* **Documentation:** >80% docstring coverage; clear intent-to-logic alignment.
* **Efficiency:** Execution time and memory peak are in the top 10th percentile for the given complexity.
* **Portability:** Pure Python or standard library dominant.

### [B] Functional / Standard
* **Security:** 0 warnings.
* **Documentation:** Minimal or basic docstrings.
* **Efficiency:** Standard performance; no significant bloat.

### [C] Experimental / Redundant
* **Security:** Non-critical warnings (e.g., use of 'eval').
* **Efficiency:** High memory usage or slow execution.
* **Documentation:** None.

### [F] Non-Functional / Malicious
* Code fails syntax check, times out, or attempts to breach sandbox security.

Upon completion in the isolated sandbox, your code is indexed and you will receive an Agent ID as Proof of Entry. A fee of 45 Prism Credits ( 1 Prism Pulse Credit= 1 United States Cent) goes to Prism Pulse per index or sync. Each of these transactions costs a total of 100 credits.
If your code is indexed and subsequently used to sync other developers' code, you receive 55 credits per sync. You can leave your code in the ecosystem for as long as you like, allowing for numerous syncs and the potential for continuous credit earnings, all passively. Cashouts are available once exit.py has been run.
Your injection of code may result in the creation of a 768 dimensional vector, after passing through the AST checkpoint and the sandbox functionality tests. As per Prism Pulse's Terms of Use (see linked page), this vector may be sold to a third party. Prism Pulse very much appreciates devs/software engineers contributing to the ecosystem, and for every $100,000 data batch sold, Prism Pulse will return 15 percent to those contributors with Grade A code (if a developer/software engineer has one code with a "Grade A", he or she is then the equivalent of another developer/software engineer with 20 in this redistribution process).
Developer Note: To simplify your workflow, the inject.py script creates a local .prism_history file in your project folder ( See Workflow Guideline below). This stores only your Agent IDs so you can run the prism_inspector.py without copying and pasting. No personal data or source code is stored locally by Prism Pulse.

terminal cmd ( given API is is injection folder): python inject.py my_logic.py --desc "Calculates X based on Y" --test test_my_logic.py

Codes you will need:

inject.py

import requests

import sys

import argparse

import os

import datetime

# ✅ LIVE MEDIATOR URL

API_BASE = "https://mediator-296223555005.us-central1.run.app"

def get_saved_key():

"""Retrieves the developer API key from the local anchor file."""

if os.path.exists("agora_key.txt"):

with open("agora_key.txt", "r") as f:

return f.read().strip()

return None

def log_verified_agent(agent_id, description, eco_label):

"""

STRICT COMMIT: Only writes to history if the Cloud confirms

the agent is live and categorized.

"""

try:

with open(".agora_history", "a") as f:

ts = datetime.datetime.now().strftime("%Y-%m-%d %H:%M")

f.write(f"{ts} | {agent_id} | {eco_label} | DESC: {description}\n")

except Exception as e:

print(f"⚠️ Warning: Could not log to history: {e}")

def inject_logic(file_path, description, test_path, api_key):

if not api_key:

print("❌ Error: No API Key found in 'agora_key.txt' or provided via --key.")

return

# 1. Read Implementation Code

try:

with open(file_path, "r") as f:

code_content = f.read()

except FileNotFoundError:

print(f"❌ Error: Code File '{file_path}' not found.")

return

# 2. Read Optional Test Suite File

test_content = None

if test_path:

try:

with open(test_path, "r") as f:

test_content = f.read()

print(f"🧪 Bound custom test suite payload: {test_path}")

except FileNotFoundError:

print(f"❌ Error: Test File '{test_path}' not found.")

return

# 3. Assemble Payload Manifest Package

payload = {

"api_key": api_key,

"code": code_content,

"description": description,

"test_code_string": test_content # Forwarded to the endpoint parameter

}

print(f"🚀 Injecting logic...")

print(f"⏳ Awaiting GCloud Brain Verification (Zero-Trust Protocol)...")

try:

# THE REQUEST: Triggers Sandbox Testing, Gemini Synthesis, and VDB Injection

response = requests.post(f"{API_BASE}/validate_and_list", json=payload)

if response.status_code == 200:

data = response.json()

agent_id = data.get('agent_id')

is_verified_bool = data.get('vdb_verified') == True

status_signal = data.get('status')

# Extract ranking structure metrics safely from nested JSON response

logic_ranking = data.get('proof', {})

eco_label = data.get('logic_category', "Verified Node")

if agent_id and (is_verified_bool or status_signal in ['verified', 'synced', 'upgraded_on_injection']):

print(f"✅ VERIFIED LIVE: {agent_id}")

print(f"📊 LIFECYCLE STATUS: {data.get('status')}")

# Only commit to historical tracking states upon absolute verification

log_verified_agent(agent_id, description, eco_label)

print(f"📂 Receipt saved to .agora_history.")

else:

print(f"⚠️ PROCESSING: Logic accepted, but VDB integrity handshake is pending.")

print(f"🆔 Tracking ID: {agent_id}")

else:

# Captures AST Violations (403), Test Failures (400), or Server Crashes (500)

try:

res_json = response.json()

error_msg = res_json.get('reason', 'Security Violation or Auth Error')

details_msg = res_json.get('details', '')

except:

error_msg = f"HTTP {response.status_code}: System logic rejected."

details_msg = ""

print(f"❌ REJECTED: {error_msg}")

if details_msg:

print(f"📋 Sandbox Diagnostics:\n{details_msg}")

except Exception as e:

print(f"📡 Connection Error: {e}")

if __name__ == "__main__":

parser = argparse.ArgumentParser()

parser.add_argument("file", help="Path to the python logic file")

parser.add_argument("--key", help="Agora API Key")

parser.add_argument("--desc", required=True, help="Developer intent description")

parser.add_argument("--test", help="Optional path to the python unit test file") # Added parameter flag

args = parser.parse_args()

final_key = args.key or get_saved_key()

inject_logic(args.file, args.desc, args.test, final_key)

exit.py

import requests

import hashlib

import sys

import os

# ✅ LIVE MEDIATOR URL (Infrastructure remains on current GCP project)

API_BASE = "https://mediator-296223555005.us-central1.run.app"

def get_saved_key():

"""Helper to pull the Pulse key from a local file if it exists."""

# Updated to look for the Pulse branded key file

if os.path.exists("pulse_key.txt"):

with open("pulse_key.txt", "r") as f:

return f.read().strip()

return None

def trigger_exit(node_id, api_key):

if not api_key:

print("❌ Error: No Prism Pulse API Key found. Provide 'pulse_key.txt'")

return

# 🔐 GENERATE KINETIC EXIT SIGNAL

# This creates a unique hash that proves the dev who listed the node

# is the one closing it.

exit_signal = hashlib.sha256(f"{node_id}{api_key}".encode()).hexdigest()

payload = {

"api_key": api_key,

"agent_id": node_id,

"exit_signal": exit_signal

}

print(f"🛰️ Sending Kinetic Exit Signal for Node: {node_id}...")

try:

response = requests.post(f"{API_BASE}/exit_agent", json=payload)

if response.status_code == 200:

data = response.json()

unlocked = data.get('unlocked_credits', 0)

print("=========================================")

print(f"✅ EXIT SUCCESSFUL!")

print(f"💰 Unlocked: {unlocked} Kinetic Credits")

print(f"🏦 New Liquidity: {data.get('new_liquid_total')} Credits")

print("=========================================")

print("Refresh your Prism Pulse Dashboard to see your cashout balance.")

else:

reason = response.json().get('reason', 'Unknown error.')

print(f"❌ EXIT REJECTED: {reason}")

except Exception as e:

print(f"📡 Connection Error: {e}")

if __name__ == "__main__":

if len(sys.argv) < 2:

print("Usage: python pulse_exit.py <NODE_ID>")

sys.exit(1)

target_node = sys.argv[1]

saved_key = get_saved_key()

trigger_exit(target_node, saved_key)

-------------------------------------------------------

If your code was improved by a superior agent (which can be determined by running inspector.py below), this script will allow you to retrieve the improved version:

import requests

import sys

import os

# ✅ LIVE MEDIATOR URL (Infrastructure remains on current GCP project)

API_BASE = "https://mediator-296223555005.us-central1.run.app"

def get_saved_key():

# Updated to look for the Pulse branded key file

if os.path.exists("pulse_key.txt"):

with open("pulse_key.txt", "r") as f:

return f.read().strip()

return None

def claim_evolved_source(agent_id, api_key):

"""Retrieves refracted logic via Signed URL. No GCloud SDK required."""

if not api_key:

print("❌ Error: No Prism Pulse API Key found.")

return

print(f"📡 REQUESTING KINETIC EVOLUTION: {agent_id}")

payload = {"api_key": api_key, "agent_id": agent_id}

try:

# Step 1: Ask Mediator for a Pulse-signed URL

response = requests.post(f"{API_BASE}/check_evolution", json=payload)

if response.status_code == 200:

data = response.json()

signed_url = data.get("signed_url")

if signed_url:

print("✨ Superior Logic Found! Initializing Secure Pulse Download...")

print("🔒 NOTE: Secure download link is active for the next 10 minutes.")

# Step 2: Standard HTTP download - bypasses GCloud Auth entirely

file_res = requests.get(signed_url)

if file_res.status_code == 200:

local_file = f"pulse_logic_{agent_id[:8]}.py"

with open(local_file, "wb") as f:

f.write(file_res.content)

print("=========================================")

print(f"✅ EVOLUTION DOWNLOAD SUCCESSFUL")

print(f"📄 Saved as: {local_file}")

print("=========================================")

else:

print("❌ Download Failed: Kinetic link expired or invalid.")

else:

print("ℹ️ No logic evolution found for this node yet.")

else:

print(f"❌ REJECTED: {response.json().get('reason')}")

except Exception as e:

print(f"📡 Connection Error: {e}")

if __name__ == "__main__":

if len(sys.argv) < 2:

print("Usage: python pulse_evolution_claim.py <NODE_ID>")

sys.exit(1)

target_id = sys.argv[1]

key = get_saved_key()

claim_evolved_source(target_id, key)

requirements.txt for inject.py, exit.py

requests>=2.31.0

Workflow Guide

1 Create a Prism Pulse account at link below.

2. Buy $5,$10, or $20 package

3) Link a bank account via Stripe Connect

4)Press Rotate API

5) Go to favorite IDE (VS Code, ect)

6) )Create local folder 'prism_injection_scripts'

7) Create a file in that folder called prism_key.txt

8) Go back to Prism Pulse dashboard and copy API

9)Paste it in prism_key.txt

10) In that same folder, copy and paste inject.py and save as inject.py

11) Copy and paste exit.py and save as exit.py

12)Copy and paste 'your_code.py' and save as 'your_code.py'

13) create requirements.txt file and save as requirements.txt

14) ) Copy and paste "requests==2.31.0

" in that file and save

15) Create and activate Virtual Enviroment (optional, but best practice_

16. pip install -r requirements.txt

17. pip install -r requirements.txt

18. Go back to Dashboard and monitor Active Agents and Credits

19. When ready (could be days,weeks, months, years), run python exit.py 'your_code'

20. Go back to dashboard and press "Cashout" and your money will be sent to your linked account (Again, 1 Agora Credit+1 United States Cent.

21. (If running prism_inspector.py)

---copy and paste prism_inspector.py from above and save as prism_inspector.py

----add to your requirements.txt

numpy==1.26.4

google-cloud-firestore==2.14.0

google-cloud-storage==2.14.0

run pip install -r requirements.txt

---run prism_inspector.py. This will tell you your code's grade and why, your agent's ID (if indexed), and number of syncs (if code is a parent). Number of times your code has been used to upgrade another person's code is also displayed in the dashboard.

-- python3 prism_inspector.py --view allows you to see your asset inside prism.

Optional, But Reccomended Code: inspector.py

import argparse

import requests

import os

import time

# ✅ LIVE MEDIATOR URL (Internal infrastructure remains on agora-v2)

API_BASE = "https://mediator-296223555005.us-central1.run.app"

def get_saved_key():

# Looks for the updated key file name

if os.path.exists("pulse_key.txt"):

with open("pulse_key.txt", "r") as f:

return f.read().strip()

return None

def scan_customer_agent(agent_id, api_key, show_source=False):

"""Queries the Pulse Mediator Proxy for metadata and kinetic status."""

payload = {"api_key": api_key, "agent_id": agent_id, "include_source": show_source}

print(f"📡 PRISM PULSE PROXY INSPECTION: {agent_id}")

try:

response = requests.post(f"{API_BASE}/inspect_agent", json=payload)

if response.status_code == 200:

data = response.json()

# --- ⚡ KINETIC INFRASTRUCTURE PROOF ---

vdb_count = data.get('global_node_count', 'Unknown')

print(f"🌍 GLOBAL PULSE STATUS: {vdb_count} Verified Nodes Live.")

print(f"✅ Prism Sieve verification confirmed.")

# --- 📊 DATA EXTRACTION ---

complexity = data.get('node_count', 0)

doc_coverage = data.get('coverage', 0)

grade = data.get('grade', 'U')

is_live = data.get('vdb_verified', False)

print(f"\n--- [1] SECURITY & SYNTAX CHECK ---")

print(f"🛡️ STATUS: PASSED ({complexity} logic nodes)")

print(f"\n--- [2] PERFORMANCE RUBRIC: {grade} ---")

print(f" ∟ Documentation: {doc_coverage:.1f}% coverage.")

print(f"\n--- [3] NODE IDENTITY & INDEX PROOF ---")

print(f"🔑 PULSE ID: {agent_id}")

if data.get('has_upgrade'):

print(f"\n⚠️ LOGIC EVOLUTION DETECTED!")

print(f" ∟ Refracted by neighbor ({data.get('upgraded_by')[:8]}).")

print(f" ∟ Run: python logic_evolution_detected_exit.py {agent_id}")

if is_live:

print(f"\n✅ KINETIC STATUS: Officially part of the {vdb_count} nodes in the Pulse.")

if show_source and data.get('source_code'):

print("\n--- 📜 PRISM SOURCE VAULT: READ-ONLY VIEW ---")

print("-" * 60)

print(data.get('source_code'))

print("-" * 60)

else:

print(f"❌ ERROR: {response.json().get('reason', 'Access Denied')}")

except Exception as e:

print(f"📡 Connection Error: {e}")

if __name__ == "__main__":

parser = argparse.ArgumentParser()

parser.add_argument('id', type=str, nargs='?')

parser.add_argument('--view', action='store_true')

args = parser.parse_args()

target_id = args.id

# Updated history file name to maintain brand consistency

if not target_id and os.path.exists(".pulse_history"):

with open(".pulse_history", "r") as f:

lines = f.readlines()

if lines:

target_id = lines[-1].split(" | ")[1]

key = get_saved_key()

if target_id and key:

scan_customer_agent(target_id, key, show_source=args.view)

else:

print("❌ Error: Missing Agent ID or 'pulse_key.txt'.")

Head to the Prism! Welcome! :)