KG edge schema reference

Module: zettelforge.knowledge_graph

from zettelforge.knowledge_graph import KnowledgeGraph, get_knowledge_graph

---

Overview

ZettelForge maintains a knowledge graph of entities and typed relationships extracted from analyst notes and detection rules. The graph stores entity nodes and directed relationship edges with optional temporal indexing for time-based queries.

Prerequisites: Use ZettelForge v2.8.0 with Python 3.10 or newer. The direct KnowledgeGraph examples on this page use local JSONL files and do not require a graph service. If you use MemoryManager, read the Memory Manager API first.

Two storage paths exist in parallel:

• SQLite (default). When you use MemoryManager with the default backend, KG writes go to the kg_nodes and kg_edges tables in zettelforge.db. This is the primary path for most deployments.
• JSONL in-process. The KnowledgeGraph class (below) writes to kg_nodes.jsonl and kg_edges.jsonl in the data directory. It is used when you call get_knowledge_graph() directly or when the TypeDB extension is not installed and the backend is explicitly set to something other than SQLite.

The edge field names and relationship types are identical in both OSS paths. The SQLite edge table adds a note_id column (see below).

This page is the OSS ZettelForge schema reference. ThreatRecall.ai SaaS uses its own tenant-scoped PostgreSQL kg_nodes and kg_edges tables with tenant_id, TLP, confidence, and SaaS-specific node and edge columns. Do not treat the OSS DDL below as the SaaS database contract; use the Web API reference for SaaS integration behavior.

TypeDB is not part of the public ZettelForge package in this checkout. Direct get_knowledge_graph() access returns JSONL unless the process is configured with a compatible graph extension. Default MemoryManager writes stay on SQLite.

Thread safety: The public add_node() and add_edge() write paths use threading.RLock.

---

Feature state

Core node and edge storage, SQLite DDL, JSONL node and edge writes, relationship labels, the functional OSINT relationships listed below, and the examples on this page were verified against ZettelForge v2.8.0 on August 18, 2026.

The direct JSONL temporal helper is in progress. In this run, one add_temporal_edge() call produced two identical get_changes_since() entries for the same change. The helper also stores timestamps verbatim and compares them as strings; it does not call _parse_timestamp() to validate or normalize them. Use consistently formatted ISO 8601 timestamps and deduplicate JSONL temporal results by (timestamp, from, relationship, to) before presenting them. The SQLite graph row shape remains verified.

---

Node schema

Nodes are created automatically when edges reference them. The entity_type field identifies the domain; the entity_value holds the canonical string.

Node fields

Field	Type	Description
node_id	str	node_<12-hex> — internal primary key
entity_type	str	Entity type string (see taxonomy below)
entity_value	str	Canonical value, e.g. apt28, CVE-2024-3094
properties	dict	Application-specific metadata
created_at	str	ISO 8601
updated_at	str	ISO 8601

Entity type taxonomy

Entity types written by each subsystem. Strings are case-sensitive and written exactly as shown.

CTI domain (written by MemoryManager._update_knowledge_graph()):

entity_type	Description
note	MemoryNote stored via remember()
actor	Threat actor via regex extraction
threat_actor	Threat actor via LLM NER
intrusion_set	Intrusion set (e.g. APT group)
tool	Malware or tool name
cve	Vulnerability identifier
asset	Target asset or sector
campaign	Campaign name
attack_pattern	MITRE ATT&CK technique (e.g. T1059)
malware	Malware family name
person	Person name (conversational domain)
location	Location (conversational domain)
organization	Organization name
event	Named event
activity	Activity name
temporal	Temporal reference extracted from text

Sigma and YARA domain (written by sigma/YARA ingest):

entity_type	Description
SigmaRule	Sigma detection rule
YaraRule	YARA detection rule
SigmaTag	Raw Sigma tag string
YaraTag	Raw YARA tag string
LogSource	Sigma logsource facet
AttackPattern	MITRE ATT&CK technique resolved from tag
Vulnerability	CVE resolved from tag
IntrusionSet	MITRE ATT&CK group resolved from sigma tag
Malware	MITRE ATT&CK software resolved from sigma tag
ThreatActor	Named threat actor resolved from YARA metadata

OSINT domain (written by functional collectors):

entity_type	Description
DomainName	Domain seed or discovered domain
IPv4Address	IPv4 address resolved from DNS, WHOIS, or BGP context
IPv6Address	IPv6 address resolved from DNS, WHOIS, or BGP context
ASNumber	Autonomous system number
Netblock	CIDR network block
MXRecord	Mail exchanger record
NSRecord	Name server record
Port	Network port discovered by the opt-in active scanner
Organization	Registrar, registrant, or owning organization
EmailAddress	Email seed or WHOIS registrant address
Alias	Username or handle used as an OSINT seed
CryptoWallet	Cryptocurrency wallet address
Transaction	Blockchain transaction
SocialAccount	Account found for an alias
Breach	Named breach associated with an email address

The OSINT module declares more ontology types than its functional collectors write. This reference lists only the OSINT nodes and relationships observed from v2.8.0 collectors and focused tests.

SQLite DDL

CREATE TABLE IF NOT EXISTS kg_nodes (
    node_id       TEXT PRIMARY KEY,
    entity_type   TEXT NOT NULL,
    entity_value  TEXT NOT NULL,
    properties    TEXT DEFAULT '{}',
    created_at    TEXT,
    updated_at    TEXT,
    UNIQUE(entity_type, entity_value)
);

---

Edge schema

Each edge is a directed relationship between two nodes.

Edge fields

Field	Type	Description
edge_id	str	edge_<12-hex> — internal primary key
from_node_id	str	Source node node_id
to_node_id	str	Target node node_id
relationship	str	Semantic relationship type (see below)
edge_type	str	heuristic | causal | detection | osint (default: heuristic)
properties	dict	Edge-specific metadata
created_at	str	ISO 8601
updated_at	str	ISO 8601

SQLite DDL

The SQLite path adds note_id, which links an edge back to the note that caused it. The unique constraint includes note_id, so the same entity pair can have the same relationship recorded from multiple notes.

CREATE TABLE IF NOT EXISTS kg_edges (
    edge_id       TEXT PRIMARY KEY,
    from_node_id  TEXT NOT NULL,
    to_node_id    TEXT NOT NULL,
    relationship  TEXT NOT NULL,
    edge_type     TEXT DEFAULT 'heuristic',
    note_id       TEXT DEFAULT '',
    properties    TEXT DEFAULT '{}',
    created_at    TEXT,
    updated_at    TEXT,
    UNIQUE(from_node_id, to_node_id, relationship, note_id)
);

The JSONL path deduplicates on (from_node_id, to_node_id, relationship) without note_id.

Edge type taxonomy

The edge_type field records how the edge was created:

edge_type	Description	Source
heuristic	Co-occurrence or heuristic extraction (default)	Entity co-occurrence during remember()
causal	LLM-extracted cause-and-effect triple	Causal triple extraction in slow-path enrichment
detection	Detection rule relationship	Sigma/YARA ingest
osint	Validated collector result	OSINT collection executor

When add_edge() is called on an existing heuristic edge with a more specific edge_type, the type is promoted to the more specific value.

---

Relationship types

CTI entity relationships

Created automatically during remember() with domain="cti".

Relationship	From	To	Description
MENTIONED_IN	any entity	note	Entity appeared in a note
USES_TOOL	actor, threat_actor, intrusion_set	tool	Actor uses a specific tool
EXPLOITS_CVE	actor, threat_actor, intrusion_set, tool	cve	Entity exploits a vulnerability
TARGETS_ASSET	actor, threat_actor, intrusion_set, tool	asset	Entity targets an asset or sector
CONDUCTS_CAMPAIGN	actor, threat_actor, intrusion_set	campaign	Actor runs a campaign
USES_TECHNIQUE	actor, threat_actor, intrusion_set	attack_pattern	Actor uses an ATT&CK technique
IMPLEMENTS	malware	attack_pattern	Malware implements a technique

Conversational entity relationships (extracted when domain is not restricted to CTI):

Relationship	From	To	Description
AFFILIATED_WITH	person	organization	Person affiliated with an org
ATTENDED	person	event	Person attended an event
LOCATED_AT	person	location	Person located at a place
PARTICIPATES_IN	person	activity	Person participates in an activity
HELD_AT	event	location	Event held at a location
ORGANIZED_BY	event	organization	Event organized by an org
OCCURRED_ON	event	temporal	Event at a temporal reference
BASED_IN	organization	location	Organization based in a location

Note versioning:

Relationship	From	To	Description
SUPERSEDES	note	note	Newer note version supersedes the older one

LLM causal relationships

Created by NoteConstructor.store_causal_edges() during slow-path enrichment. These edges have edge_type="causal". Extraction requires a configured LLM provider.

Relationship	Description
causes	Subject is a direct cause of object
enables	Subject enables or facilitates object
targets	Subject targets object
uses	Subject uses object
exploits	Subject exploits object
attributed_to	Subject is attributed to object
related_to	Generic causal relationship

The relationship value is the lowercase string as validated against CAUSAL_RELATIONS (note_constructor.py:98). Node entity_type and entity_value for causal edges are free-form strings extracted by the LLM and are not constrained to the entity type taxonomy above.

Detection rule relationships

Created during Sigma and YARA ingest.

Relationship	From	To	Description
applies_to	SigmaRule	LogSource	Rule applies to a log source facet
tagged_with	SigmaRule, YaraRule	SigmaTag, YaraTag	Rule carries a tag
detects	SigmaRule, YaraRule	AttackPattern	Rule detects an ATT&CK technique
references_cve	SigmaRule, YaraRule	Vulnerability	Rule references a CVE
attributed_to	SigmaRule, YaraRule	IntrusionSet, Malware, ThreatActor	Rule attributed to a group or actor
superseded_by	SigmaRule	SigmaRule	Rule replaced by a newer version
related_to	SigmaRule	SigmaRule	Generic rule relationship

OSINT relationships

Created by functional zettelforge.osint collectors through run_osint_collection(). Passive collectors write through the normal KG API. The port scanner writes only when you explicitly enable active scanning with ZETTELFORGE_OSINT_ACTIVE_SCAN=1.

Relationship	From	To	Description
resolves_to	DomainName	IPv4Address, IPv6Address	DNS A or AAAA result
ns_for	DomainName	NSRecord	DNS NS record
mx_for	DomainName	MXRecord	DNS MX record
owned_by	DomainName, Netblock	Organization	WHOIS registrant or netblock owner
associated_with	IPv4Address, IPv6Address	Netblock	IP address belongs to a netblock
part_of_as	IPv4Address, IPv6Address, Netblock	ASNumber	IP address or netblock is announced by an autonomous system
listens_on	IPv4Address, IPv6Address	Port	Active scanner found an open port
related_to	DomainName	DomainName	Certificate transparency SAN domain association
registered_by	DomainName	EmailAddress	WHOIS data names a registrant email address
has_account	Alias	SocialAccount	Account discovery links a handle to a platform account
appeared_in_breach	EmailAddress	Breach	Breach lookup associates an email address with a named breach
sent_transaction	CryptoWallet	Transaction	Wallet appears as the transaction sender
received_transaction	CryptoWallet	Transaction	Wallet appears as the transaction recipient

The OSINT ontology also declares hosts, delegated_to, receives_mail_on, and controls_wallet. No v2.8.0 functional collector emitted those relationships during this audit, so treat them as declared vocabulary, not observed writes.

Temporal relationships

For tracking entity state over time. Used with add_temporal_edge().

Relationship	Description
TEMPORAL_BEFORE	State or event occurred before another
TEMPORAL_AFTER	State or event occurred after another
SUPERSEDES	New entity state supersedes an old one

Temporal edges are indexed automatically in _temporal_index and _entity_timeline on write.

---

Legacy schema normalization

Edges written by pre-v2.5.1 deployments used different key names. _normalize_edge_schema() rewrites them on load:

Legacy key	Canonical key
source_id	from_node_id
target_id	to_node_id
relation_type	relationship

Rules for rejected entries:

• Entry missing edge_id: dropped.
• Entry missing from_node_id, to_node_id, or relationship (and no recoverable legacy keys): dropped.
• Malformed JSON line: skipped; total count logged at warning level as kg_edges_skipped_malformed.

This was a production hotfix (v2.5.1) for long-running deployments where pre-v2.5.x writers had written ~80k+ legacy entries alongside canonical-shape rows in kg_edges.jsonl.

---

KnowledgeGraph class API

Use KnowledgeGraph directly when you need in-process JSONL graph access, or call get_knowledge_graph() for the shared singleton.

add_node()

def add_node(self, entity_type: str, entity_value: str, properties: dict | None = None) -> str:
    ...

Creates or updates a node. Returns node_id. If the node already exists, merges properties and refreshes updated_at.

add_edge()

def add_edge(
    self,
    from_type: str,
    from_value: str,
    to_type: str,
    to_value: str,
    relationship: str,
    properties: dict | None = None,
) -> str:
    ...

Creates or updates a directed edge. Auto-creates both endpoint nodes. Deduplicates on (from_node_id, to_node_id, relationship). If a duplicate exists with edge_type="heuristic" and you pass a more specific edge_type in properties, it is promoted.

get_node()

def get_node(self, entity_type: str, entity_value: str) -> dict | None:
    ...

Looks up a node by type and value. Returns None if not found.

get_node_by_id()

def get_node_by_id(self, node_id: str) -> dict | None:
    ...

Looks up a node by its internal node_id.

get_outgoing_edges()

def get_outgoing_edges(self, node_id: str) -> list[dict]:
    ...

Returns all outgoing edges for a node_id. Each dict contains at minimum edge_id, from_node_id, to_node_id, relationship, properties, created_at, updated_at.

get_neighbors()

def get_neighbors(
    self, entity_type: str, entity_value: str, relationship: str | None = None
) -> list[dict]:
    ...

Returns adjacent nodes reachable via outgoing edges, with optional relationship filter. Each result contains node (the target node dict), relationship, and edge_properties.

traverse()

def traverse(self, start_type: str, start_value: str, max_depth: int = 2) -> list[dict]:
    ...

Depth-first traversal up to max_depth. Returns a list of paths; each path is a list of step dicts with keys from_type, from_value, relationship, to_type, to_value.

---

Temporal indexing

add_temporal_edge()

def add_temporal_edge(
    self,
    from_type: str,
    from_value: str,
    to_type: str,
    to_value: str,
    relationship: str,    # TEMPORAL_BEFORE | TEMPORAL_AFTER | SUPERSEDES
    timestamp: str,
    properties: dict | None = None,
) -> str:
    ...

Creates an edge and indexes it in _temporal_index (keyed by timestamp string) and _entity_timeline (keyed by "entity_type:entity_value").

Timestamp handling

add_temporal_edge() stores the timestamp string you pass without validation. get_changes_since() filters and sorts those strings lexicographically. Use one fixed-width ISO 8601 representation, such as 2026-01-15T00:00:00Z, throughout a graph so string order matches chronological order.

The private _parse_timestamp() helper can parse the following formats, but the public temporal write and query methods do not call it in v2.8.0:

• ISO 8601 (including Z suffix)
• %Y-%m-%d
• %Y-%m-%d %H:%M:%S
• %d %b %Y (e.g. 15 Jan 2026)
• %B %d, %Y (e.g. January 15, 2026)

get_entity_timeline()

def get_entity_timeline(self, entity_type: str, entity_value: str) -> list[dict]:
    ...

Returns the chronological state history for an entity. Each entry: {"edge": dict, "timestamp": str, "to_entity": "type:value"}.

get_changes_since()

def get_changes_since(self, timestamp: str) -> list[dict]:
    ...

Returns temporal entity changes whose stored timestamp string is lexicographically greater than or equal to timestamp. Results are sorted by that string. Each entry: {"timestamp": str, "from": "type:value", "relationship": str, "to": "type:value"}.

get_latest_state()

def get_latest_state(self, entity_type: str, entity_value: str) -> dict | None:
    ...

Returns the most recent temporal edge entry for an entity, or None if no temporal data exists.

---

Causal edge queries

Causal edges (edge_type: "causal") represent LLM-extracted cause-and-effect triples. Two dedicated methods traverse them.

get_causal_edges()

def get_causal_edges(
    self, entity_type: str, entity_value: str,
    max_depth: int = 3, max_visited: int = 50,
) -> list[dict]:
    ...

BFS over outgoing causal edges — traces forward from cause to effects. Useful for "what does this actor or event lead to?" queries.

get_incoming_causal()

def get_incoming_causal(
    self, entity_type: str, entity_value: str,
    max_depth: int = 3, max_visited: int = 50,
) -> list[dict]:
    ...

BFS over incoming causal edges — traces back to root causes. Useful for "why did this happen?" queries.

---

Global singleton

from zettelforge.knowledge_graph import KnowledgeGraph


def get_knowledge_graph() -> KnowledgeGraph:
    ...

Returns the process-global KnowledgeGraph instance for direct graph access. Checks ZETTELFORGE_BACKEND (default: sqlite):

• If ZETTELFORGE_BACKEND=typedb and a compatible TypeDB graph extension is importable, it tries that extension.
• Otherwise, it returns the JSONL KnowledgeGraph instance.

When you use MemoryManager with the default SQLite backend, KG writes go directly to the SQLite tables via self.store.add_kg_edge(), not through get_knowledge_graph(). Call get_knowledge_graph() for direct graph access independent of the memory manager.

---

Examples

from zettelforge.knowledge_graph import KnowledgeGraph

kg = KnowledgeGraph()

# Add nodes and an edge
kg.add_node("actor", "apt28")
kg.add_node("tool", "cobalt strike")
edge_id = kg.add_edge("actor", "apt28", "tool", "cobalt strike", "USES_TOOL")

# Query neighbors
neighbors = kg.get_neighbors("actor", "apt28")
for n in neighbors:
    print(f"{n['relationship']} -> {n['node']['entity_type']}:{n['node']['entity_value']}")
# Output: USES_TOOL -> tool:cobalt strike

# Traverse up to depth 2
paths = kg.traverse("actor", "apt28", max_depth=2)
for path in paths:
    print(" -> ".join(
        f"{p['from_type']}:{p['from_value']} [{p['relationship']}] {p['to_type']}:{p['to_value']}"
        for p in path
    ))

from zettelforge.knowledge_graph import KnowledgeGraph

kg = KnowledgeGraph()

# Temporal edge: record that an event preceded another
kg.add_temporal_edge(
    from_type="actor", from_value="apt28",
    to_type="campaign", to_value="nato-phishing-2026-q1",
    relationship="TEMPORAL_BEFORE",
    timestamp="2026-01-15",
)

# What changed since the start of Q1?
changes = kg.get_changes_since("2026-01-01")
for c in changes:
    print(f"[{c['timestamp']}] {c['from']} {c['relationship']} {c['to']}")

# Causal chain: forward (cause → effects)
causes = kg.get_causal_edges("actor", "apt28", max_depth=3)

# Causal chain: backward (root cause analysis)
root_causes = kg.get_incoming_causal("campaign", "nato-phishing-2026-q1", max_depth=3)