memory/memory.py · Cyclops V2

203 lines · python

"""
Cyclops v2 - unified memory layer.
 
Three stores, each with a distinct job (no overlap, all actually used):
  - KuzuDB graph  : structural knowledge + state. Targets, archetypes, sinks,
                    dataflow hypotheses, oracle proofs, findings, and CLEARED AXES.
                    This is what stops drift/tunnel-vision: the agent queries
                    "what have I already cleared on this target?" before acting.
  - Chroma        : semantic recall over prior reasoning / methods / failed attempts.
  - Redis         : working memory - the live job queue + content-addressed analysis
                    cache (never re-CPG the same commit) + per-run scratch.
 
Design rule (anti-slop): a FINDING node may only be created once an oracle PROOF
node with a real execution artifact is attached. A dataflow HYPOTHESIS is not a
finding. memory.promote_to_finding() enforces this.
"""
import os, sys, json, time, hashlib
import kuzu
import redis as redislib
 
CYCLOPS_HOME = os.environ.get("CYCLOPS_HOME") or os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
GRAPH_PATH   = os.path.join(CYCLOPS_HOME, "memory", "graph.kuzu")
REDIS_URL    = os.environ.get("CYCLOPS_REDIS", "redis://127.0.0.1:6379/3")
 
SCHEMA = [
    """CREATE NODE TABLE IF NOT EXISTS Archetype(
         name STRING, description STRING, PRIMARY KEY(name))""",
    """CREATE NODE TABLE IF NOT EXISTS Target(
         id STRING, repo STRING, archetype STRING, lang STRING,
         commit STRING, released_at STRING, active BOOLEAN, status STRING,
         first_seen STRING, PRIMARY KEY(id))""",
    """CREATE NODE TABLE IF NOT EXISTS SinkClass(
         name STRING, cwe STRING, impact_family STRING, PRIMARY KEY(name))""",
    """CREATE NODE TABLE IF NOT EXISTS Hypothesis(
         id STRING, target STRING, sink_class STRING, file STRING, line INT64,
         source STRING, sink STRING, tool STRING, created_at STRING,
         status STRING, PRIMARY KEY(id))""",
    """CREATE NODE TABLE IF NOT EXISTS Proof(
         id STRING, hypothesis STRING, label STRING, oracle STRING,
         artifact_path STRING, summary STRING, created_at STRING,
         PRIMARY KEY(id))""",
    """CREATE NODE TABLE IF NOT EXISTS Finding(
         id STRING, target STRING, impact_family STRING, severity STRING,
         title STRING, novel BOOLEAN, cve_checked BOOLEAN, created_at STRING,
         PRIMARY KEY(id))""",
    """CREATE NODE TABLE IF NOT EXISTS ClearedAxis(
         id STRING, target STRING, archetype STRING, axis STRING,
         result STRING, reason STRING, created_at STRING, PRIMARY KEY(id))""",
    "CREATE REL TABLE IF NOT EXISTS OF_ARCHETYPE(FROM Target TO Archetype)",
    "CREATE REL TABLE IF NOT EXISTS HAS_HYPOTHESIS(FROM Target TO Hypothesis)",
    "CREATE REL TABLE IF NOT EXISTS TARGETS_SINK(FROM Hypothesis TO SinkClass)",
    "CREATE REL TABLE IF NOT EXISTS PROVEN_BY(FROM Hypothesis TO Proof)",
    "CREATE REL TABLE IF NOT EXISTS YIELDED(FROM Proof TO Finding)",
    "CREATE REL TABLE IF NOT EXISTS CLEARED_ON(FROM Target TO ClearedAxis)",
]
 
def _now():
    return time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
 
def _hash(*parts):
    return hashlib.sha1("|".join(str(p) for p in parts).encode()).hexdigest()[:16]
 
 
class Memory:
    def __init__(self):
        self.db = kuzu.Database(GRAPH_PATH)
        self.g = kuzu.Connection(self.db)
        for stmt in SCHEMA:
            self.g.execute(stmt)
        self.r = redislib.from_url(REDIS_URL, decode_responses=True)
 
    def _exec(self, q, params=None):
        return self.g.execute(q, parameters=params or {})
 
    def upsert_archetype(self, name, description=""):
        self._exec("MERGE (a:Archetype {name:$n}) SET a.description=$d",
                   {"n": name, "d": description})
 
    def upsert_target(self, repo, archetype, lang="", commit="", released_at="",
                      active=True, status="DISCOVERED"):
        tid = _hash(repo, commit)
        self._exec("""MERGE (t:Target {id:$id})
                      SET t.repo=$repo, t.archetype=$arch, t.lang=$lang,
                          t.commit=$commit, t.released_at=$rel, t.active=$active,
                          t.status=$status, t.first_seen=coalesce(t.first_seen,$now)""",
                   {"id": tid, "repo": repo, "arch": archetype, "lang": lang,
                    "commit": commit, "rel": released_at, "active": active,
                    "status": status, "now": _now()})
        self.upsert_archetype(archetype)
        self._exec("""MATCH (t:Target {id:$id}),(a:Archetype {name:$arch})
                      MERGE (t)-[:OF_ARCHETYPE]->(a)""", {"id": tid, "arch": archetype})
        return tid
 
    def add_hypothesis(self, target_id, sink_class, file, line, source, sink, tool):
        hid = _hash(target_id, file, line, sink, source)
        self._exec("""MERGE (h:Hypothesis {id:$id})
                      SET h.target=$t, h.sink_class=$sc, h.file=$f, h.line=$ln,
                          h.source=$src, h.sink=$snk, h.tool=$tool,
                          h.created_at=$now, h.status='REACHABLE'""",
                   {"id": hid, "t": target_id, "sc": sink_class, "f": file,
                    "ln": int(line), "src": source, "snk": sink, "tool": tool,
                    "now": _now()})
        self._exec("""MATCH (t:Target {id:$t}),(h:Hypothesis {id:$id})
                      MERGE (t)-[:HAS_HYPOTHESIS]->(h)""", {"t": target_id, "id": hid})
        return hid
 
    def add_proof(self, hypothesis_id, label, oracle, artifact_path, summary):
        """label in {SOURCE_CONFIRMED, REPLICA_CONFIRMED, LIVE_CANARY_CONFIRMED, DISPROVEN}"""
        pid = _hash(hypothesis_id, label, oracle, artifact_path)
        self._exec("""MERGE (p:Proof {id:$id})
                      SET p.hypothesis=$h, p.label=$lbl, p.oracle=$o,
                          p.artifact_path=$ap, p.summary=$s, p.created_at=$now""",
                   {"id": pid, "h": hypothesis_id, "lbl": label, "o": oracle,
                    "ap": artifact_path, "s": summary, "now": _now()})
        self._exec("""MATCH (h:Hypothesis {id:$h}),(p:Proof {id:$id})
                      MERGE (h)-[:PROVEN_BY]->(p)
                      SET h.status=$st""",
                   {"h": hypothesis_id, "id": pid,
                    "st": "DISPROVEN" if label == "DISPROVEN" else "PROVEN"})
        return pid
 
    def promote_to_finding(self, hypothesis_id, impact_family, severity, title,
                           novel=False, cve_checked=False):
        """ANTI-SLOP GATE: refuses unless a non-DISPROVEN Proof exists."""
        res = self._exec("""MATCH (h:Hypothesis {id:$h})-[:PROVEN_BY]->(p:Proof)
                            WHERE p.label <> 'DISPROVEN'
                            RETURN p.id, h.target LIMIT 1""", {"h": hypothesis_id})
        if not res.has_next():
            raise PermissionError(
                f"REFUSED: hypothesis {hypothesis_id} has no execution proof. "
                "A reachable dataflow path is NOT a finding. Run the oracle first.")
        row = res.get_next()
        pid, target = row[0], row[1]
        fid = _hash(hypothesis_id, impact_family, title)
        self._exec("""MERGE (f:Finding {id:$id})
                      SET f.target=$t, f.impact_family=$fam, f.severity=$sev,
                          f.title=$title, f.novel=$novel, f.cve_checked=$cc,
                          f.created_at=$now""",
                   {"id": fid, "t": target, "fam": impact_family, "sev": severity,
                    "title": title, "novel": novel, "cc": cve_checked, "now": _now()})
        self._exec("""MATCH (p:Proof {id:$p}),(f:Finding {id:$f})
                      MERGE (p)-[:YIELDED]->(f)""", {"p": pid, "f": fid})
        return fid
 
    def clear_axis(self, target_id, archetype, axis, result, reason=""):
        aid = _hash(target_id, axis)
        self._exec("""MERGE (c:ClearedAxis {id:$id})
                      SET c.target=$t, c.archetype=$arch, c.axis=$axis,
                          c.result=$res, c.reason=$reason, c.created_at=$now""",
                   {"id": aid, "t": target_id, "arch": archetype, "axis": axis,
                    "res": result, "reason": reason, "now": _now()})
        self._exec("""MATCH (t:Target {id:$t}),(c:ClearedAxis {id:$id})
                      MERGE (t)-[:CLEARED_ON]->(c)""", {"t": target_id, "id": aid})
        return aid
 
    def cleared_axes(self, target_id):
        res = self._exec("""MATCH (t:Target {id:$t})-[:CLEARED_ON]->(c:ClearedAxis)
                            RETURN c.axis, c.result, c.reason ORDER BY c.created_at""",
                         {"t": target_id})
        out = []
        while res.has_next():
            r = res.get_next(); out.append({"axis": r[0], "result": r[1], "reason": r[2]})
        return out
 
    def is_axis_cleared(self, target_id, axis):
        res = self._exec("""MATCH (t:Target {id:$t})-[:CLEARED_ON]->(c:ClearedAxis {axis:$a})
                            RETURN c.result LIMIT 1""", {"t": target_id, "a": axis})
        return res.get_next()[0] if res.has_next() else None
 
    def cache_get(self, key):
        return self.r.get(f"cyclops:cache:{key}")
 
    def cache_put(self, key, value, ttl=86400 * 30):
        self.r.set(f"cyclops:cache:{key}", value, ex=ttl)
 
    def analysis_done(self, target_id, phase):
        """content-addressed: never re-run the same phase on the same commit."""
        return self.r.sismember("cyclops:analysis_done", f"{target_id}:{phase}")
 
    def mark_analysis_done(self, target_id, phase):
        self.r.sadd("cyclops:analysis_done", f"{target_id}:{phase}")
 
    def enqueue(self, queue, job: dict):
        self.r.lpush(f"cyclops:q:{queue}", json.dumps(job))
 
    def dequeue(self, queue, timeout=5):
        item = self.r.brpop(f"cyclops:q:{queue}", timeout=timeout)
        return json.loads(item[1]) if item else None
 
    def stats(self):
        res = self._exec("""MATCH (n) RETURN label(n) AS l, count(*) AS c""")
        out = {}
        while res.has_next():
            r = res.get_next(); out[r[0]] = r[1]
        return out
 
 
if __name__ == "__main__":
    m = Memory()
    print("Cyclops v2 memory online.")
    print("graph:", GRAPH_PATH)
    print("redis:", REDIS_URL, "ping=", m.r.ping())
    print("node counts:", m.stats())

1	"""
2	Cyclops v2 - unified memory layer.
3
4	Three stores, each with a distinct job (no overlap, all actually used):
5	- KuzuDB graph : structural knowledge + state. Targets, archetypes, sinks,
6	dataflow hypotheses, oracle proofs, findings, and CLEARED AXES.
7	This is what stops drift/tunnel-vision: the agent queries
8	"what have I already cleared on this target?" before acting.
9	- Chroma : semantic recall over prior reasoning / methods / failed attempts.
10	- Redis : working memory - the live job queue + content-addressed analysis
11	cache (never re-CPG the same commit) + per-run scratch.
12
13	Design rule (anti-slop): a FINDING node may only be created once an oracle PROOF
14	node with a real execution artifact is attached. A dataflow HYPOTHESIS is not a
15	finding. memory.promote_to_finding() enforces this.
16	"""
17	import os, sys, json, time, hashlib
18	import kuzu
19	import redis as redislib
20
21	CYCLOPS_HOME = os.environ.get("CYCLOPS_HOME") or os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
22	GRAPH_PATH = os.path.join(CYCLOPS_HOME, "memory", "graph.kuzu")
23	REDIS_URL = os.environ.get("CYCLOPS_REDIS", "redis://127.0.0.1:6379/3")
24
25	SCHEMA = [
26	"""CREATE NODE TABLE IF NOT EXISTS Archetype(
27	name STRING, description STRING, PRIMARY KEY(name))""",
28	"""CREATE NODE TABLE IF NOT EXISTS Target(
29	id STRING, repo STRING, archetype STRING, lang STRING,
30	commit STRING, released_at STRING, active BOOLEAN, status STRING,
31	first_seen STRING, PRIMARY KEY(id))""",
32	"""CREATE NODE TABLE IF NOT EXISTS SinkClass(
33	name STRING, cwe STRING, impact_family STRING, PRIMARY KEY(name))""",
34	"""CREATE NODE TABLE IF NOT EXISTS Hypothesis(
35	id STRING, target STRING, sink_class STRING, file STRING, line INT64,
36	source STRING, sink STRING, tool STRING, created_at STRING,
37	status STRING, PRIMARY KEY(id))""",
38	"""CREATE NODE TABLE IF NOT EXISTS Proof(
39	id STRING, hypothesis STRING, label STRING, oracle STRING,
40	artifact_path STRING, summary STRING, created_at STRING,
41	PRIMARY KEY(id))""",
42	"""CREATE NODE TABLE IF NOT EXISTS Finding(
43	id STRING, target STRING, impact_family STRING, severity STRING,
44	title STRING, novel BOOLEAN, cve_checked BOOLEAN, created_at STRING,
45	PRIMARY KEY(id))""",
46	"""CREATE NODE TABLE IF NOT EXISTS ClearedAxis(
47	id STRING, target STRING, archetype STRING, axis STRING,
48	result STRING, reason STRING, created_at STRING, PRIMARY KEY(id))""",
49	"CREATE REL TABLE IF NOT EXISTS OF_ARCHETYPE(FROM Target TO Archetype)",
50	"CREATE REL TABLE IF NOT EXISTS HAS_HYPOTHESIS(FROM Target TO Hypothesis)",
51	"CREATE REL TABLE IF NOT EXISTS TARGETS_SINK(FROM Hypothesis TO SinkClass)",
52	"CREATE REL TABLE IF NOT EXISTS PROVEN_BY(FROM Hypothesis TO Proof)",
53	"CREATE REL TABLE IF NOT EXISTS YIELDED(FROM Proof TO Finding)",
54	"CREATE REL TABLE IF NOT EXISTS CLEARED_ON(FROM Target TO ClearedAxis)",
55	]
56
57	def _now():
58	return time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())
59
60	def _hash(*parts):
61	return hashlib.sha1("\|".join(str(p) for p in parts).encode()).hexdigest()[:16]
62
63
64	class Memory:
65	def __init__(self):
66	self.db = kuzu.Database(GRAPH_PATH)
67	self.g = kuzu.Connection(self.db)
68	for stmt in SCHEMA:
69	self.g.execute(stmt)
70	self.r = redislib.from_url(REDIS_URL, decode_responses=True)
71
72	def _exec(self, q, params=None):
73	return self.g.execute(q, parameters=params or {})
74
75	def upsert_archetype(self, name, description=""):
76	self._exec("MERGE (a:Archetype {name:$n}) SET a.description=$d",
77	{"n": name, "d": description})
78
79	def upsert_target(self, repo, archetype, lang="", commit="", released_at="",
80	active=True, status="DISCOVERED"):
81	tid = _hash(repo, commit)
82	self._exec("""MERGE (t:Target {id:$id})
83	SET t.repo=$repo, t.archetype=$arch, t.lang=$lang,
84	t.commit=$commit, t.released_at=$rel, t.active=$active,
85	t.status=$status, t.first_seen=coalesce(t.first_seen,$now)""",
86	{"id": tid, "repo": repo, "arch": archetype, "lang": lang,
87	"commit": commit, "rel": released_at, "active": active,
88	"status": status, "now": _now()})
89	self.upsert_archetype(archetype)
90	self._exec("""MATCH (t:Target {id:$id}),(a:Archetype {name:$arch})
91	MERGE (t)-[:OF_ARCHETYPE]->(a)""", {"id": tid, "arch": archetype})
92	return tid
93
94	def add_hypothesis(self, target_id, sink_class, file, line, source, sink, tool):
95	hid = _hash(target_id, file, line, sink, source)
96	self._exec("""MERGE (h:Hypothesis {id:$id})
97	SET h.target=$t, h.sink_class=$sc, h.file=$f, h.line=$ln,
98	h.source=$src, h.sink=$snk, h.tool=$tool,
99	h.created_at=$now, h.status='REACHABLE'""",
100	{"id": hid, "t": target_id, "sc": sink_class, "f": file,
101	"ln": int(line), "src": source, "snk": sink, "tool": tool,
102	"now": _now()})
103	self._exec("""MATCH (t:Target {id:$t}),(h:Hypothesis {id:$id})
104	MERGE (t)-[:HAS_HYPOTHESIS]->(h)""", {"t": target_id, "id": hid})
105	return hid
106
107	def add_proof(self, hypothesis_id, label, oracle, artifact_path, summary):
108	"""label in {SOURCE_CONFIRMED, REPLICA_CONFIRMED, LIVE_CANARY_CONFIRMED, DISPROVEN}"""
109	pid = _hash(hypothesis_id, label, oracle, artifact_path)
110	self._exec("""MERGE (p:Proof {id:$id})
111	SET p.hypothesis=$h, p.label=$lbl, p.oracle=$o,
112	p.artifact_path=$ap, p.summary=$s, p.created_at=$now""",
113	{"id": pid, "h": hypothesis_id, "lbl": label, "o": oracle,
114	"ap": artifact_path, "s": summary, "now": _now()})
115	self._exec("""MATCH (h:Hypothesis {id:$h}),(p:Proof {id:$id})
116	MERGE (h)-[:PROVEN_BY]->(p)
117	SET h.status=$st""",
118	{"h": hypothesis_id, "id": pid,
119	"st": "DISPROVEN" if label == "DISPROVEN" else "PROVEN"})
120	return pid
121
122	def promote_to_finding(self, hypothesis_id, impact_family, severity, title,
123	novel=False, cve_checked=False):
124	"""ANTI-SLOP GATE: refuses unless a non-DISPROVEN Proof exists."""
125	res = self._exec("""MATCH (h:Hypothesis {id:$h})-[:PROVEN_BY]->(p:Proof)
126	WHERE p.label <> 'DISPROVEN'
127	RETURN p.id, h.target LIMIT 1""", {"h": hypothesis_id})
128	if not res.has_next():
129	raise PermissionError(
130	f"REFUSED: hypothesis {hypothesis_id} has no execution proof. "
131	"A reachable dataflow path is NOT a finding. Run the oracle first.")
132	row = res.get_next()
133	pid, target = row[0], row[1]
134	fid = _hash(hypothesis_id, impact_family, title)
135	self._exec("""MERGE (f:Finding {id:$id})
136	SET f.target=$t, f.impact_family=$fam, f.severity=$sev,
137	f.title=$title, f.novel=$novel, f.cve_checked=$cc,
138	f.created_at=$now""",
139	{"id": fid, "t": target, "fam": impact_family, "sev": severity,
140	"title": title, "novel": novel, "cc": cve_checked, "now": _now()})
141	self._exec("""MATCH (p:Proof {id:$p}),(f:Finding {id:$f})
142	MERGE (p)-[:YIELDED]->(f)""", {"p": pid, "f": fid})
143	return fid
144
145	def clear_axis(self, target_id, archetype, axis, result, reason=""):
146	aid = _hash(target_id, axis)
147	self._exec("""MERGE (c:ClearedAxis {id:$id})
148	SET c.target=$t, c.archetype=$arch, c.axis=$axis,
149	c.result=$res, c.reason=$reason, c.created_at=$now""",
150	{"id": aid, "t": target_id, "arch": archetype, "axis": axis,
151	"res": result, "reason": reason, "now": _now()})
152	self._exec("""MATCH (t:Target {id:$t}),(c:ClearedAxis {id:$id})
153	MERGE (t)-[:CLEARED_ON]->(c)""", {"t": target_id, "id": aid})
154	return aid
155
156	def cleared_axes(self, target_id):
157	res = self._exec("""MATCH (t:Target {id:$t})-[:CLEARED_ON]->(c:ClearedAxis)
158	RETURN c.axis, c.result, c.reason ORDER BY c.created_at""",
159	{"t": target_id})
160	out = []
161	while res.has_next():
162	r = res.get_next(); out.append({"axis": r[0], "result": r[1], "reason": r[2]})
163	return out
164
165	def is_axis_cleared(self, target_id, axis):
166	res = self._exec("""MATCH (t:Target {id:$t})-[:CLEARED_ON]->(c:ClearedAxis {axis:$a})
167	RETURN c.result LIMIT 1""", {"t": target_id, "a": axis})
168	return res.get_next()[0] if res.has_next() else None
169
170	def cache_get(self, key):
171	return self.r.get(f"cyclops:cache:{key}")
172
173	def cache_put(self, key, value, ttl=86400 * 30):
174	self.r.set(f"cyclops:cache:{key}", value, ex=ttl)
175
176	def analysis_done(self, target_id, phase):
177	"""content-addressed: never re-run the same phase on the same commit."""
178	return self.r.sismember("cyclops:analysis_done", f"{target_id}:{phase}")
179
180	def mark_analysis_done(self, target_id, phase):
181	self.r.sadd("cyclops:analysis_done", f"{target_id}:{phase}")
182
183	def enqueue(self, queue, job: dict):
184	self.r.lpush(f"cyclops:q:{queue}", json.dumps(job))
185
186	def dequeue(self, queue, timeout=5):
187	item = self.r.brpop(f"cyclops:q:{queue}", timeout=timeout)
188	return json.loads(item[1]) if item else None
189
190	def stats(self):
191	res = self._exec("""MATCH (n) RETURN label(n) AS l, count(*) AS c""")
192	out = {}
193	while res.has_next():
194	r = res.get_next(); out[r[0]] = r[1]
195	return out
196
197
198	if __name__ == "__main__":
199	m = Memory()
200	print("Cyclops v2 memory online.")
201	print("graph:", GRAPH_PATH)
202	print("redis:", REDIS_URL, "ping=", m.r.ping())
203	print("node counts:", m.stats())