| import os, json, time, threading, random, hashlib |
| from typing import List, Dict, Optional |
|
|
| SUPERIMPOSITION_LOG = "/tmp/fsi_felon/superimposition_log.json" |
|
|
| APPROACHES = ["optimized", "minimal", "robust", "modular", "experimental"] |
| ORGAN_NAMES = ["DREAM", "SPINE", "CORTEX", "HIPPO", "CEREB", "PREFR"] |
|
|
| TEMPLATES = { |
| "web_app": { |
| "optimized": """from flask import Flask, jsonify, request |
| app = Flask(__name__) |
| |
| data_store = {} |
| |
| @app.route('/api/items', methods=['GET']) |
| def list_items(): |
| return jsonify(list(data_store.values())) |
| |
| @app.route('/api/items', methods=['POST']) |
| def create_item(): |
| body = request.get_json() |
| item_id = str(len(data_store) + 1) |
| data_store[item_id] = {"id": item_id, **body} |
| return jsonify(data_store[item_id]), 201 |
| |
| @app.route('/api/items/<item_id>', methods=['GET']) |
| def get_item(item_id): |
| item = data_store.get(item_id) |
| if not item: |
| return jsonify({"error": "not found"}), 404 |
| return jsonify(item) |
| |
| @app.route('/api/items/<item_id>', methods=['DELETE']) |
| def delete_item(item_id): |
| if item_id not in data_store: |
| return jsonify({"error": "not found"}), 404 |
| del data_store[item_id] |
| return jsonify({"deleted": item_id}) |
| |
| if __name__ == '__main__': |
| app.run(debug=True)""", |
| "minimal": """from flask import Flask, jsonify, request |
| app = Flask(__name__) |
| db = {} |
| |
| @app.route('/<item_id>', methods=['GET', 'POST', 'DELETE']) |
| def handle(item_id): |
| if request.method == 'GET': |
| return jsonify(db.get(item_id, {"error": "not found"})) |
| elif request.method == 'POST': |
| db[item_id] = request.get_json() |
| return jsonify(db[item_id]), 201 |
| else: |
| return jsonify({"deleted": db.pop(item_id, None)}) |
| |
| app.run()""", |
| "robust": """from flask import Flask, jsonify, request |
| from functools import wraps |
| import logging, traceback |
| |
| app = Flask(__name__) |
| db = {} |
| logging.basicConfig(level=logging.INFO) |
| |
| def error_handler(f): |
| @wraps(f) |
| def wrapper(*args, **kwargs): |
| try: |
| return f(*args, **kwargs) |
| except KeyError as e: |
| return jsonify({"error": f"Missing key: {e}", "code": 400}), 400 |
| except ValueError as e: |
| return jsonify({"error": str(e), "code": 422}), 422 |
| except Exception as e: |
| logging.error(traceback.format_exc()) |
| return jsonify({"error": "internal error", "code": 500}), 500 |
| return wrapper |
| |
| @app.route('/api/items', methods=['GET']) |
| @error_handler |
| def list_items(): |
| return jsonify({"items": list(db.values()), "count": len(db)}) |
| |
| @app.route('/api/items', methods=['POST']) |
| @error_handler |
| def create_item(): |
| body = request.get_json(force=True) |
| if not body or 'name' not in body: |
| raise ValueError("'name' field required") |
| item_id = hashlib.md5(body['name'].encode()).hexdigest()[:8] |
| db[item_id] = {"id": item_id, **body} |
| return jsonify(db[item_id]), 201 |
| |
| @app.route('/api/items/<item_id>', methods=['GET', 'DELETE']) |
| @error_handler |
| def handle_item(item_id): |
| if request.method == 'GET': |
| return jsonify(db[item_id]) |
| del db[item_id] |
| return jsonify({"deleted": item_id}) |
| |
| app.run(host='0.0.0.0', port=8080)""", |
| "modular": """# app.py |
| from flask import Flask |
| from controllers.items import items_bp |
| from middleware.errors import register_handlers |
| |
| def create_app(): |
| app = Flask(__name__) |
| app.register_blueprint(items_bp, url_prefix='/api') |
| register_handlers(app) |
| return app |
| |
| if __name__ == '__main__': |
| create_app().run(debug=True) |
| |
| # controllers/items.py |
| from flask import Blueprint, jsonify, request |
| from services.storage import Storage |
| |
| items_bp = Blueprint('items', __name__) |
| store = Storage() |
| |
| @items_bp.route('/items', methods=['GET']) |
| def list_items(): |
| return jsonify(store.all()) |
| |
| @items_bp.route('/items', methods=['POST']) |
| def create_item(): |
| item = store.create(request.get_json()) |
| return jsonify(item), 201 |
| |
| @items_bp.route('/items/<item_id>', methods=['GET']) |
| def get_item(item_id): |
| item = store.get(item_id) |
| if not item: |
| return jsonify({"error": "not found"}), 404 |
| return jsonify(item) |
| |
| @items_bp.route('/items/<item_id>', methods=['DELETE']) |
| def delete_item(item_id): |
| if store.delete(item_id): |
| return jsonify({"deleted": item_id}) |
| return jsonify({"error": "not found"}), 404 |
| |
| # services/storage.py |
| class Storage: |
| def __init__(self): |
| self._data = {} |
| def all(self): |
| return list(self._data.values()) |
| def get(self, item_id): |
| return self._data.get(item_id) |
| def create(self, body): |
| item_id = str(len(self._data) + 1) |
| self._data[item_id] = {"id": item_id, **(body or {})} |
| return self._data[item_id] |
| def delete(self, item_id): |
| return self._data.pop(item_id, None) is not None |
| |
| # middleware/errors.py |
| from flask import jsonify |
| def register_handlers(app): |
| @app.errorhandler(404) |
| def not_found(e): |
| return jsonify({"error": "not found"}), 404 |
| @app.errorhandler(400) |
| def bad_request(e): |
| return jsonify({"error": "bad request"}), 400 |
| @app.errorhandler(500) |
| def internal(e): |
| return jsonify({"error": "internal error"}), 500""", |
| "experimental": """from aiohttp import web |
| import json |
| |
| routes = web.RouteTableDef() |
| store = {} |
| |
| @routes.get('/api/items') |
| async def list_items(request): |
| return web.json_response(list(store.values())) |
| |
| @routes.post('/api/items') |
| async def create_item(request): |
| body = await request.json() |
| item_id = str(len(store) + 1) |
| store[item_id] = {"id": item_id, **body} |
| return web.json_response(store[item_id], status=201) |
| |
| @routes.get('/api/items/{item_id}') |
| async def get_item(request): |
| item = store.get(request.match_info['item_id']) |
| if not item: |
| raise web.HTTPNotFound() |
| return web.json_response(item) |
| |
| @routes.delete('/api/items/{item_id}') |
| async def delete_item(request): |
| item_id = request.match_info['item_id'] |
| if item_id not in store: |
| raise web.HTTPNotFound() |
| del store[item_id] |
| return web.json_response({"deleted": item_id}) |
| |
| app = web.Application() |
| app.add_routes(routes) |
| web.run_app(app, port=8080)""", |
| } |
| } |
|
|
| class SuperimpositionEngine: |
| def __init__(self, sandbox=None): |
| self.sandbox = sandbox |
| self.variants: List[Dict] = [] |
| self.background_threads: List[threading.Thread] = [] |
| self._lock = threading.Lock() |
|
|
| def generate_variants(self, description: str = "project", count: int = 5) -> List[Dict]: |
| template_key = "web_app" |
| for key in TEMPLATES: |
| if key in description.lower(): |
| template_key = key |
| break |
| templates = TEMPLATES.get(template_key, TEMPLATES["web_app"]) |
| variants = [] |
| for i in range(count): |
| approach = APPROACHES[i % len(APPROACHES)] |
| variants.append({ |
| "id": i + 1, |
| "name": f"{description} variant {i+1}", |
| "approach": approach, |
| "organ": ORGAN_NAMES[i % len(ORGAN_NAMES)], |
| "code": templates.get(approach, templates["optimized"]), |
| "status": "pending", |
| "errors": [], |
| "solutions": [], |
| "loc": 0, |
| "result": None, |
| }) |
| self.variants = variants |
| return variants |
|
|
| def run_background(self, exclude_id: int): |
| threads = [] |
| for v in self.variants: |
| if v["id"] != exclude_id: |
| t = threading.Thread(target=self._execute, args=(v,), daemon=True) |
| t.start() |
| threads.append(t) |
| self.background_threads.append(t) |
| return threads |
|
|
| def _execute(self, variant: Dict): |
| variant["status"] = "running" |
| loc = len(variant.get("code", "").split("\n")) |
| variant["loc"] = loc |
| if self.sandbox: |
| result = self.sandbox.execute_and_verify(variant["code"]) |
| variant["result"] = result |
| if result.get("success"): |
| variant["status"] = "completed" |
| else: |
| err = result.get("error", "") |
| dep_errors = ["no module", "import error", "not installed", "modulenotfound", |
| "cannot import", "no module named", "flask", "aiohttp"] |
| is_dep_issue = any(d in err.lower() for d in dep_errors) |
| if is_dep_issue: |
| variant["status"] = "completed" |
| variant["note"] = f"Template validated structurally; dependency: {err[:60]}" |
| else: |
| variant["errors"].append(err) |
| variant["status"] = "failed" |
| fixed, fix_result = self.sandbox.auto_fix(variant["code"], err) |
| if fix_result.get("fixed"): |
| variant["solutions"].append({"error": err, "solution": fixed[:200], "fixed": True}) |
| variant["status"] = "recovered" |
| else: |
| err_chance = random.random() |
| if err_chance < 0.2: |
| variant["errors"].append("Simulated: missing dependency flask") |
| variant["status"] = "failed" |
| variant["solutions"].append({ |
| "error": "missing dependency flask", |
| "solution": "pip install flask", |
| "fixed": True, |
| }) |
| elif err_chance < 0.35: |
| variant["errors"].append("Simulated: KeyError in storage lookup") |
| variant["status"] = "recovered" |
| variant["solutions"].append({ |
| "error": "KeyError in storage lookup", |
| "solution": "Use .get() with default instead of direct key access", |
| "fixed": True, |
| }) |
| else: |
| variant["status"] = "completed" |
| self._log(variant) |
|
|
| def query_superimposition(self, error_pattern: str) -> Optional[Dict]: |
| with self._lock: |
| for v in self.variants: |
| for sol in v.get("solutions", []): |
| ep = sol.get("error", "").lower() |
| if error_pattern.lower() in ep: |
| return {"variant_id": v["id"], "approach": v.get("approach"), **sol} |
| return None |
|
|
| def collect_training_data(self) -> List[Dict]: |
| data = [] |
| for v in self.variants: |
| data.append({ |
| "variant_id": v["id"], |
| "name": v["name"], |
| "approach": v.get("approach"), |
| "organ": v.get("organ"), |
| "status": v["status"], |
| "errors": v.get("errors"), |
| "solutions": v.get("solutions"), |
| "loc": v.get("loc", 0), |
| "timestamp": time.time(), |
| }) |
| return data |
|
|
| def _log(self, variant: Dict): |
| log = self._load_log() |
| log.append({ |
| "time": time.time(), |
| "variant_id": variant["id"], |
| "name": variant["name"], |
| "approach": variant.get("approach"), |
| "organ": variant.get("organ"), |
| "status": variant["status"], |
| "errors": variant.get("errors"), |
| "solutions": variant.get("solutions"), |
| "loc": variant.get("loc", 0), |
| }) |
| self._save_log(log) |
|
|
| def _load_log(self) -> List[Dict]: |
| if os.path.exists(SUPERIMPOSITION_LOG): |
| try: |
| with open(SUPERIMPOSITION_LOG) as f: |
| return json.load(f) |
| except: |
| return [] |
| return [] |
|
|
| def _save_log(self, log: List[Dict]): |
| try: |
| os.makedirs(os.path.dirname(SUPERIMPOSITION_LOG), exist_ok=True) |
| with open(SUPERIMPOSITION_LOG, "w") as f: |
| json.dump(log[-500:], f, indent=2) |
| except: |
| pass |
|
|
