Skip to content

naqsha.memory

Dynamic Memory Engine and Memory Port adapters

naqsha.memory

Memory Port adapters and Dynamic Memory Engine.

MemoryPort

Bases: Protocol

Durable memory boundary owned by the Core Runtime.

Source code in src/naqsha/memory/base.py 
17
18
19
20
21
22
23
24
25
26
27
28
29
30
class MemoryPort(Protocol):
    """Durable memory boundary owned by the Core Runtime."""

    def start_run(self, run_id: str, query: str) -> None:
        """Start memory lifecycle for a run."""

    def retrieve(self, query: str, token_budget: int) -> list[MemoryRecord]:
        """Retrieve provenance-aware memory records."""

    def record_observation(self, run_id: str, tool: str, observation: ToolObservation) -> None:
        """Record sanitized tool observations."""

    def finish_run(self, run_id: str, answer: str | None) -> None:
        """Finalize memory lifecycle for a run."""

start_run 

start_run(run_id: str, query: str) -> None

Start memory lifecycle for a run.

Source code in src/naqsha/memory/base.py 
20
21
def start_run(self, run_id: str, query: str) -> None:
    """Start memory lifecycle for a run."""

retrieve 

retrieve(
    query: str, token_budget: int
) -> list[MemoryRecord]

Retrieve provenance-aware memory records.

Source code in src/naqsha/memory/base.py 
23
24
def retrieve(self, query: str, token_budget: int) -> list[MemoryRecord]:
    """Retrieve provenance-aware memory records."""

record_observation 

record_observation(
    run_id: str, tool: str, observation: ToolObservation
) -> None

Record sanitized tool observations.

Source code in src/naqsha/memory/base.py 
26
27
def record_observation(self, run_id: str, tool: str, observation: ToolObservation) -> None:
    """Record sanitized tool observations."""

finish_run 

finish_run(run_id: str, answer: str | None) -> None

Finalize memory lifecycle for a run.

Source code in src/naqsha/memory/base.py 
29
30
def finish_run(self, run_id: str, answer: str | None) -> None:
    """Finalize memory lifecycle for a run."""

MemoryRecord dataclass

Source code in src/naqsha/memory/base.py 
11
12
13
14
@dataclass(frozen=True)
class MemoryRecord:
    content: str
    provenance: str

content instance-attribute 

content: str

provenance instance-attribute 

provenance: str

ForbiddenDDLError

Bases: Exception

Raised when DDL statement violates the safelist.

Source code in src/naqsha/memory/ddl.py 
30
31
32
33
class ForbiddenDDLError(Exception):
    """Raised when DDL statement violates the safelist."""

    pass

DynamicMemoryEngine

SQLite-backed memory engine with shared and private namespaces.

The engine manages a single SQLite database with namespace-isolated tables: - Shared tables: prefixed with shared_ - Private tables: prefixed with private_<agent_id>_

Optional sqlite-vec support for semantic search can be enabled via config.

Source code in src/naqsha/memory/engine.py 
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
class DynamicMemoryEngine:
    """SQLite-backed memory engine with shared and private namespaces.

    The engine manages a single SQLite database with namespace-isolated tables:
    - Shared tables: prefixed with `shared_`
    - Private tables: prefixed with `private_<agent_id>_`

    Optional sqlite-vec support for semantic search can be enabled via config.
    """

    def __init__(
        self,
        *,
        db_path: Path | str,
        enable_embeddings: bool = False,
    ) -> None:
        """Initialize the Dynamic Memory Engine.

        Args:
            db_path: Path to SQLite database file
            enable_embeddings: Whether to load sqlite-vec for embeddings
        """
        self._db_path = Path(db_path)
        self._enable_embeddings = enable_embeddings

        # Ensure parent directory exists
        self._db_path.parent.mkdir(parents=True, exist_ok=True)

        # Open connection
        self._conn = sqlite3.connect(
            str(self._db_path), isolation_level=None, check_same_thread=False
        )
        self._conn.row_factory = sqlite3.Row

        # Enable WAL mode for better concurrency
        self._conn.execute("PRAGMA journal_mode=WAL")

        # Load sqlite-vec if embeddings are enabled
        if self._enable_embeddings:
            self._load_sqlite_vec()

    def _load_sqlite_vec(self) -> None:
        """Load sqlite-vec extension for vector embeddings.

        Raises:
            ImportError: If sqlite-vec is not installed
        """
        try:
            import sqlite_vec  # type: ignore

            # Load the extension
            sqlite_vec.load(self._conn)
        except ImportError as e:
            raise ImportError(
                "sqlite-vec is required for embeddings support. "
                "Install with: pip install naqsha[embeddings]"
            ) from e

    def close(self) -> None:
        """Close the database connection."""
        self._conn.close()

    def get_shared_scope(self) -> MemoryScope:
        """Get a MemoryScope for shared (team-wide) memory.

        Returns:
            MemoryScope with 'shared_' namespace
        """
        return MemoryScope(self._conn, namespace="shared_", agent_id=None)

    def get_private_scope(self, agent_id: str) -> MemoryScope:
        """Get a MemoryScope for private (agent-specific) memory.

        Args:
            agent_id: Agent identifier

        Returns:
            MemoryScope with 'private_<agent_id>_' namespace
        """
        if not agent_id:
            raise ValueError("agent_id cannot be empty")

        namespace = f"private_{agent_id}_"
        return MemoryScope(self._conn, namespace=namespace, agent_id=agent_id)

    @property
    def connection(self) -> sqlite3.Connection:
        """Get the underlying SQLite connection.

        This is exposed for advanced use cases but should generally not be used directly.
        Use get_shared_scope() or get_private_scope() instead.
        """
        return self._conn

    @property
    def embeddings_enabled(self) -> bool:
        """Check if embeddings are enabled."""
        return self._enable_embeddings

    def list_all_tables(self) -> dict[str, list[str]]:
        """List all tables grouped by namespace.

        Returns:
            Dictionary mapping namespace to list of table names (without prefix)
        """
        cursor = self._conn.execute(
            """
            SELECT name FROM sqlite_master
            WHERE type='table' AND name NOT LIKE 'sqlite_%'
            ORDER BY name
            """
        )

        tables: dict[str, list[str]] = {
            "shared": [],
            "private": {},
        }

        for row in cursor:
            name = row[0]
            if name.startswith("shared_"):
                tables["shared"].append(name[7:])  # Remove 'shared_' prefix
            elif name.startswith("private_"):
                # Extract agent_id from 'private_<agent_id>_<table>'
                rest = name[8:]  # Remove 'private_' prefix
                if "_" in rest:
                    agent_id, table_name = rest.split("_", 1)
                    if agent_id not in tables["private"]:
                        tables["private"][agent_id] = []
                    tables["private"][agent_id].append(table_name)

        return tables

connection property 

connection: Connection

Get the underlying SQLite connection.

This is exposed for advanced use cases but should generally not be used directly. Use get_shared_scope() or get_private_scope() instead.

embeddings_enabled property 

embeddings_enabled: bool

Check if embeddings are enabled.

close 

close() -> None

Close the database connection.

Source code in src/naqsha/memory/engine.py 
77
78
79
def close(self) -> None:
    """Close the database connection."""
    self._conn.close()

get_shared_scope 

get_shared_scope() -> MemoryScope

Get a MemoryScope for shared (team-wide) memory.

Returns:

Type Description
MemoryScope

MemoryScope with 'shared_' namespace
Source code in src/naqsha/memory/engine.py 
81
82
83
84
85
86
87
def get_shared_scope(self) -> MemoryScope:
    """Get a MemoryScope for shared (team-wide) memory.

    Returns:
        MemoryScope with 'shared_' namespace
    """
    return MemoryScope(self._conn, namespace="shared_", agent_id=None)

get_private_scope 

get_private_scope(agent_id: str) -> MemoryScope

Get a MemoryScope for private (agent-specific) memory.

Parameters:

Name Type Description Default
agent_id str

Agent identifier

 required

Returns:

Type Description
MemoryScope

MemoryScope with 'private__' namespace
Source code in src/naqsha/memory/engine.py 
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
def get_private_scope(self, agent_id: str) -> MemoryScope:
    """Get a MemoryScope for private (agent-specific) memory.

    Args:
        agent_id: Agent identifier

    Returns:
        MemoryScope with 'private_<agent_id>_' namespace
    """
    if not agent_id:
        raise ValueError("agent_id cannot be empty")

    namespace = f"private_{agent_id}_"
    return MemoryScope(self._conn, namespace=namespace, agent_id=agent_id)

list_all_tables 

list_all_tables() -> dict[str, list[str]]

List all tables grouped by namespace.

Returns:

Type Description
dict[str, list[str]]

Dictionary mapping namespace to list of table names (without prefix)
Source code in src/naqsha/memory/engine.py 
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
def list_all_tables(self) -> dict[str, list[str]]:
    """List all tables grouped by namespace.

    Returns:
        Dictionary mapping namespace to list of table names (without prefix)
    """
    cursor = self._conn.execute(
        """
        SELECT name FROM sqlite_master
        WHERE type='table' AND name NOT LIKE 'sqlite_%'
        ORDER BY name
        """
    )

    tables: dict[str, list[str]] = {
        "shared": [],
        "private": {},
    }

    for row in cursor:
        name = row[0]
        if name.startswith("shared_"):
            tables["shared"].append(name[7:])  # Remove 'shared_' prefix
        elif name.startswith("private_"):
            # Extract agent_id from 'private_<agent_id>_<table>'
            rest = name[8:]  # Remove 'private_' prefix
            if "_" in rest:
                agent_id, table_name = rest.split("_", 1)
                if agent_id not in tables["private"]:
                    tables["private"][agent_id] = []
                tables["private"][agent_id].append(table_name)

    return tables

InMemoryMemoryPort

Source code in src/naqsha/memory/inmemory.py 
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
class InMemoryMemoryPort:
    def __init__(self) -> None:
        self.records: list[MemoryRecord] = []
        self.started_runs: list[str] = []
        self.finished_runs: list[str] = []

    def start_run(self, run_id: str, query: str) -> None:
        self.started_runs.append(run_id)

    def retrieve(self, query: str, token_budget: int) -> list[MemoryRecord]:
        cap = max(0, token_budget) * 4
        picked: list[MemoryRecord] = []
        spent = 0
        for rec in reversed(self.records):
            chunk = len(rec.content)
            if spent + chunk > cap and picked:
                break
            if spent + chunk > cap and not picked:
                body = rec.content[: max(0, cap - spent)]
                if not body.strip():
                    break
                picked.append(MemoryRecord(content=body, provenance=rec.provenance))
                spent = cap
                break
            picked.append(rec)
            spent += chunk
        return list(reversed(picked))

    def record_observation(self, run_id: str, tool: str, observation: ToolObservation) -> None:
        if observation.ok:
            self.records.append(
                MemoryRecord(content=observation.content, provenance=f"{run_id}:{tool}")
            )

    def finish_run(self, run_id: str, answer: str | None) -> None:
        self.finished_runs.append(run_id)

records instance-attribute 

records: list[MemoryRecord] = []

started_runs instance-attribute 

started_runs: list[str] = []

finished_runs instance-attribute 

finished_runs: list[str] = []

start_run 

start_run(run_id: str, query: str) -> None
Source code in src/naqsha/memory/inmemory.py 
15
16
def start_run(self, run_id: str, query: str) -> None:
    self.started_runs.append(run_id)

retrieve 

retrieve(
    query: str, token_budget: int
) -> list[MemoryRecord]
Source code in src/naqsha/memory/inmemory.py 
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
def retrieve(self, query: str, token_budget: int) -> list[MemoryRecord]:
    cap = max(0, token_budget) * 4
    picked: list[MemoryRecord] = []
    spent = 0
    for rec in reversed(self.records):
        chunk = len(rec.content)
        if spent + chunk > cap and picked:
            break
        if spent + chunk > cap and not picked:
            body = rec.content[: max(0, cap - spent)]
            if not body.strip():
                break
            picked.append(MemoryRecord(content=body, provenance=rec.provenance))
            spent = cap
            break
        picked.append(rec)
        spent += chunk
    return list(reversed(picked))

record_observation 

record_observation(
    run_id: str, tool: str, observation: ToolObservation
) -> None
Source code in src/naqsha/memory/inmemory.py 
37
38
39
40
41
def record_observation(self, run_id: str, tool: str, observation: ToolObservation) -> None:
    if observation.ok:
        self.records.append(
            MemoryRecord(content=observation.content, provenance=f"{run_id}:{tool}")
        )

finish_run 

finish_run(run_id: str, answer: str | None) -> None
Source code in src/naqsha/memory/inmemory.py 
43
44
def finish_run(self, run_id: str, answer: str | None) -> None:
    self.finished_runs.append(run_id)

MemoryRetriever

Token-budgeted memory retrieval with keyword and recency ranking.

Source code in src/naqsha/memory/retrieval.py 
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
class MemoryRetriever:
    """Token-budgeted memory retrieval with keyword and recency ranking."""

    def __init__(
        self,
        scope: MemoryScope,
        *,
        enable_semantic: bool = False,
    ) -> None:
        """Initialize memory retriever.

        Args:
            scope: MemoryScope to retrieve from
            enable_semantic: Whether to use semantic (embedding) ranking
        """
        self._scope = scope
        self._enable_semantic = enable_semantic

    def retrieve(
        self,
        query: str,
        token_budget: int,
        *,
        table_name: str = "memories",
        content_column: str = "content",
        provenance_column: str = "provenance",
        timestamp_column: str = "created_ts",
    ) -> list[str]:
        """Retrieve memory records within token budget.

        Args:
            query: Query text for relevance ranking
            token_budget: Maximum tokens to return
            table_name: Name of memory table (without namespace prefix)
            content_column: Name of content column
            provenance_column: Name of provenance column
            timestamp_column: Name of timestamp column

        Returns:
            List of provenance-wrapped memory strings
        """
        chars_left = _approx_chars_for_tokens(token_budget)
        if chars_left == 0:
            return []

        # Retrieve candidate records
        candidates = self._retrieve_candidates(
            query=query,
            table_name=table_name,
            content_column=content_column,
            provenance_column=provenance_column,
            timestamp_column=timestamp_column,
            limit=512,
        )

        # Rank by keyword match + recency
        ranked = self._rank_candidates(query, candidates)

        # Pack into token budget
        return self._pack_within_budget(ranked, chars_left)

    def _retrieve_candidates(
        self,
        query: str,
        table_name: str,
        content_column: str,
        provenance_column: str,
        timestamp_column: str,
        limit: int,
    ) -> list[RankedMemoryRecord]:
        """Retrieve candidate records from database.

        Returns records ordered by recency (most recent first).
        """
        try:
            rows = self._scope.query(
                f"""
                SELECT {content_column}, {provenance_column}, {timestamp_column}
                FROM {table_name}
                ORDER BY {timestamp_column} DESC
                LIMIT ?
                """,
                (limit,),
            )
        except Exception:
            # Table might not exist yet
            return []

        candidates = []
        for row in rows:
            candidates.append(
                RankedMemoryRecord(
                    content=row[content_column],
                    provenance=row[provenance_column],
                    created_ts=float(row[timestamp_column]),
                    rank_score=0.0,  # Will be computed in ranking
                )
            )

        return candidates

    def _rank_candidates(
        self,
        query: str,
        candidates: list[RankedMemoryRecord],
    ) -> list[RankedMemoryRecord]:
        """Rank candidates by keyword match + recency.

        Ranking formula: (keyword_hits * 1000000) + created_ts
        This prioritizes keyword matches strongly while using recency as a tiebreaker.
        The large multiplier ensures keyword hits dominate over timestamp differences.
        """
        tokens = _tokenize(query)

        ranked = []
        for record in candidates:
            if tokens:
                haystack = record.content.lower()
                hits = sum(1 for tok in tokens if _haystack_matches_token(haystack, tok))
                # Use large multiplier to ensure keyword hits dominate
                score = (hits * 1000000.0) + record.created_ts
            else:
                # No query tokens, rank by recency only
                score = record.created_ts

            ranked.append(
                RankedMemoryRecord(
                    content=record.content,
                    provenance=record.provenance,
                    created_ts=record.created_ts,
                    rank_score=score,
                )
            )

        # Sort by rank score (highest first)
        ranked.sort(key=lambda r: r.rank_score, reverse=True)
        return ranked

    def _pack_within_budget(
        self,
        ranked: list[RankedMemoryRecord],
        chars_left: int,
    ) -> list[str]:
        """Pack ranked records into token budget.

        Returns provenance-wrapped memory strings.
        """
        results: list[str] = []
        seen_provenance: set[str] = set()

        for record in ranked:
            # Skip duplicates
            if record.provenance in seen_provenance:
                continue

            seen_provenance.add(record.provenance)

            # Wrap with provenance
            wrapped = _wrap_memory_evidence(record.provenance, record.content)

            # Check if it fits
            if len(wrapped) > chars_left:
                if results:
                    # Already have some results, stop here
                    break

                # First record must fit, trim if necessary
                overhead = len(wrapped) - len(record.content)
                body_budget = max(0, chars_left - overhead)
                trimmed_content = record.content.strip()[:body_budget].rstrip()

                if not trimmed_content:
                    # Can't fit anything
                    break

                wrapped = _wrap_memory_evidence(record.provenance, trimmed_content)

                if len(wrapped) > chars_left:
                    # Still doesn't fit even after trimming
                    break

            results.append(wrapped)
            chars_left -= len(wrapped)

            if chars_left <= 0:
                break

        return results

retrieve 

retrieve(
    query: str,
    token_budget: int,
    *,
    table_name: str = "memories",
    content_column: str = "content",
    provenance_column: str = "provenance",
    timestamp_column: str = "created_ts",
) -> list[str]

Retrieve memory records within token budget.

Parameters:

Name Type Description Default
query str

Query text for relevance ranking

 required
token_budget int

Maximum tokens to return

 required
table_name str

Name of memory table (without namespace prefix)

 'memories'
content_column str

Name of content column

 'content'
provenance_column str

Name of provenance column

 'provenance'
timestamp_column str

Name of timestamp column

 'created_ts'

Returns:

Type Description
list[str]

List of provenance-wrapped memory strings
Source code in src/naqsha/memory/retrieval.py 
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
def retrieve(
    self,
    query: str,
    token_budget: int,
    *,
    table_name: str = "memories",
    content_column: str = "content",
    provenance_column: str = "provenance",
    timestamp_column: str = "created_ts",
) -> list[str]:
    """Retrieve memory records within token budget.

    Args:
        query: Query text for relevance ranking
        token_budget: Maximum tokens to return
        table_name: Name of memory table (without namespace prefix)
        content_column: Name of content column
        provenance_column: Name of provenance column
        timestamp_column: Name of timestamp column

    Returns:
        List of provenance-wrapped memory strings
    """
    chars_left = _approx_chars_for_tokens(token_budget)
    if chars_left == 0:
        return []

    # Retrieve candidate records
    candidates = self._retrieve_candidates(
        query=query,
        table_name=table_name,
        content_column=content_column,
        provenance_column=provenance_column,
        timestamp_column=timestamp_column,
        limit=512,
    )

    # Rank by keyword match + recency
    ranked = self._rank_candidates(query, candidates)

    # Pack into token budget
    return self._pack_within_budget(ranked, chars_left)

MemoryScope

Scoped access to memory with namespace enforcement.

A MemoryScope provides SQL execution with automatic namespace prefix enforcement. All table names are automatically prefixed with the scope's namespace.

Source code in src/naqsha/memory/scope.py 
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
class MemoryScope:
    """Scoped access to memory with namespace enforcement.

    A MemoryScope provides SQL execution with automatic namespace prefix enforcement.
    All table names are automatically prefixed with the scope's namespace.
    """

    def __init__(
        self,
        conn: sqlite3.Connection,
        namespace: str,
        agent_id: str | None = None,
    ) -> None:
        """Initialize a memory scope.

        Args:
            conn: SQLite connection
            namespace: Namespace prefix (e.g., "shared_" or "private_agent1_")
            agent_id: Agent ID for private scopes (None for shared)
        """
        self._conn = conn
        self._namespace = namespace
        self._agent_id = agent_id

        # Validate namespace format
        if not namespace.endswith("_"):
            raise ValueError(f"Namespace must end with underscore: {namespace}")

        if namespace.startswith("private_"):
            if not agent_id:
                raise ValueError("Private namespace requires agent_id")
            expected = f"private_{agent_id}_"
            if namespace != expected:
                raise ValueError(
                    f"Private namespace mismatch: expected {expected}, got {namespace}"
                )
        elif namespace != "shared_":
            raise ValueError(
                f"Invalid namespace: must be 'shared_' or 'private_<agent_id>_', got {namespace}"
            )

    @property
    def namespace(self) -> str:
        """Get the namespace prefix."""
        return self._namespace

    @property
    def agent_id(self) -> str | None:
        """Get the agent ID (None for shared scope)."""
        return self._agent_id

    def _prefix_table_names(self, sql: str) -> str:
        """Add namespace prefix to table names in SQL statement.

        This is a simple implementation that handles common cases.
        It looks for table names after CREATE TABLE, FROM, JOIN, INTO, UPDATE, etc.
        """
        # For DDL statements, prefix the table name
        if is_ddl_statement(sql):
            # CREATE TABLE table_name -> CREATE TABLE namespace_table_name
            sql = re.sub(
                r"(CREATE\s+TABLE\s+(?:IF\s+NOT\s+EXISTS\s+)?)" + r"(\w+)",
                rf"\1{self._namespace}\2",
                sql,
                flags=re.IGNORECASE,
            )
            # CREATE INDEX ... ON table_name -> CREATE INDEX ... ON namespace_table_name
            sql = re.sub(
                r"(\bON\s+)(\w+)",
                rf"\1{self._namespace}\2",
                sql,
                flags=re.IGNORECASE,
            )
            # ALTER TABLE table_name -> ALTER TABLE namespace_table_name
            sql = re.sub(
                r"(ALTER\s+TABLE\s+)(\w+)",
                rf"\1{self._namespace}\2",
                sql,
                flags=re.IGNORECASE,
            )
        else:
            # For DML statements, prefix table names after FROM, JOIN, INTO, UPDATE
            # FROM table_name
            sql = re.sub(
                r"(\bFROM\s+)(\w+)",
                rf"\1{self._namespace}\2",
                sql,
                flags=re.IGNORECASE,
            )
            # JOIN table_name
            sql = re.sub(
                r"(\bJOIN\s+)(\w+)",
                rf"\1{self._namespace}\2",
                sql,
                flags=re.IGNORECASE,
            )
            # INTO table_name
            sql = re.sub(
                r"(\bINTO\s+)(\w+)",
                rf"\1{self._namespace}\2",
                sql,
                flags=re.IGNORECASE,
            )
            # UPDATE table_name
            sql = re.sub(
                r"(UPDATE\s+)(\w+)",
                rf"\1{self._namespace}\2",
                sql,
                flags=re.IGNORECASE,
            )

        return sql

    def execute(
        self,
        sql: str,
        params: tuple[Any, ...] | dict[str, Any] | None = None,
    ) -> sqlite3.Cursor:
        """Execute SQL with namespace enforcement.

        Args:
            sql: SQL statement
            params: Query parameters

        Returns:
            SQLite cursor

        Raises:
            ForbiddenDDLError: If DDL statement violates safelist
        """
        # Validate DDL if applicable (only for schema-changing operations)
        if is_ddl_statement(sql):
            validate_ddl(sql)

        # Prefix table names with namespace
        prefixed_sql = self._prefix_table_names(sql)

        if params is None:
            return self._conn.execute(prefixed_sql)
        else:
            return self._conn.execute(prefixed_sql, params)

    def query(
        self,
        sql: str,
        params: tuple[Any, ...] | dict[str, Any] | None = None,
    ) -> list[sqlite3.Row]:
        """Execute query and return all rows.

        Args:
            sql: SQL query
            params: Query parameters

        Returns:
            List of rows
        """
        cursor = self.execute(sql, params)
        return cursor.fetchall()

    def begin(self) -> None:
        """Begin a transaction."""
        self._conn.execute("BEGIN")

    def commit(self) -> None:
        """Commit the current transaction."""
        self._conn.commit()

    def rollback(self) -> None:
        """Rollback the current transaction."""
        self._conn.rollback()

    def list_tables(self) -> list[str]:
        """List all tables in this namespace.

        Returns:
            List of table names (without namespace prefix)
        """
        cursor = self._conn.execute(
            """
            SELECT name FROM sqlite_master
            WHERE type='table' AND name LIKE ?
            ORDER BY name
            """,
            (f"{self._namespace}%",),
        )
        tables = []
        for row in cursor:
            name = row[0]
            # Remove namespace prefix
            if name.startswith(self._namespace):
                tables.append(name[len(self._namespace) :])
        return tables

namespace property 

namespace: str

Get the namespace prefix.

agent_id property 

agent_id: str | None

Get the agent ID (None for shared scope).

execute 

execute(
    sql: str,
    params: tuple[Any, ...] | dict[str, Any] | None = None,
) -> sqlite3.Cursor

Execute SQL with namespace enforcement.

Parameters:

Name Type Description Default
sql str

SQL statement

 required
params tuple[Any, ...] | dict[str, Any] | None

Query parameters

 None

Returns:

Type Description
Cursor

SQLite cursor

Raises:

Type Description
ForbiddenDDLError

If DDL statement violates safelist
Source code in src/naqsha/memory/scope.py 
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
def execute(
    self,
    sql: str,
    params: tuple[Any, ...] | dict[str, Any] | None = None,
) -> sqlite3.Cursor:
    """Execute SQL with namespace enforcement.

    Args:
        sql: SQL statement
        params: Query parameters

    Returns:
        SQLite cursor

    Raises:
        ForbiddenDDLError: If DDL statement violates safelist
    """
    # Validate DDL if applicable (only for schema-changing operations)
    if is_ddl_statement(sql):
        validate_ddl(sql)

    # Prefix table names with namespace
    prefixed_sql = self._prefix_table_names(sql)

    if params is None:
        return self._conn.execute(prefixed_sql)
    else:
        return self._conn.execute(prefixed_sql, params)

query 

query(
    sql: str,
    params: tuple[Any, ...] | dict[str, Any] | None = None,
) -> list[sqlite3.Row]

Execute query and return all rows.

Parameters:

Name Type Description Default
sql str

SQL query

 required
params tuple[Any, ...] | dict[str, Any] | None

Query parameters

 None

Returns:

Type Description
list[Row]

List of rows
Source code in src/naqsha/memory/scope.py 
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
def query(
    self,
    sql: str,
    params: tuple[Any, ...] | dict[str, Any] | None = None,
) -> list[sqlite3.Row]:
    """Execute query and return all rows.

    Args:
        sql: SQL query
        params: Query parameters

    Returns:
        List of rows
    """
    cursor = self.execute(sql, params)
    return cursor.fetchall()

begin 

begin() -> None

Begin a transaction.

Source code in src/naqsha/memory/scope.py 
174
175
176
def begin(self) -> None:
    """Begin a transaction."""
    self._conn.execute("BEGIN")

commit 

commit() -> None

Commit the current transaction.

Source code in src/naqsha/memory/scope.py 
178
179
180
def commit(self) -> None:
    """Commit the current transaction."""
    self._conn.commit()

rollback 

rollback() -> None

Rollback the current transaction.

Source code in src/naqsha/memory/scope.py 
182
183
184
def rollback(self) -> None:
    """Rollback the current transaction."""
    self._conn.rollback()

list_tables 

list_tables() -> list[str]

List all tables in this namespace.

Returns:

Type Description
list[str]

List of table names (without namespace prefix)
Source code in src/naqsha/memory/scope.py 
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
def list_tables(self) -> list[str]:
    """List all tables in this namespace.

    Returns:
        List of table names (without namespace prefix)
    """
    cursor = self._conn.execute(
        """
        SELECT name FROM sqlite_master
        WHERE type='table' AND name LIKE ?
        ORDER BY name
        """,
        (f"{self._namespace}%",),
    )
    tables = []
    for row in cursor:
        name = row[0]
        # Remove namespace prefix
        if name.startswith(self._namespace):
            tables.append(name[len(self._namespace) :])
    return tables

TeamMemoryConfig dataclass

Memory section from naqsha.toml.

Source code in src/naqsha/memory/sharing.py 
15
16
17
18
19
20
21
22
23
24
25
26
27
@dataclass(frozen=True)
class TeamMemoryConfig:
    """Memory section from ``naqsha.toml``."""

    type: str = "sqlite"
    db_path: Path = Path(".naqsha/memory.db")
    embeddings: bool = False

    def resolve_paths(self, base_dir: Path) -> TeamMemoryConfig:
        path = self.db_path
        if not path.is_absolute():
            path = (base_dir / path).resolve()
        return TeamMemoryConfig(type=self.type, db_path=path, embeddings=self.embeddings)

type class-attribute instance-attribute 

type: str = 'sqlite'

db_path class-attribute instance-attribute 

db_path: Path = Path('.naqsha/memory.db')

embeddings class-attribute instance-attribute 

embeddings: bool = False

resolve_paths 

resolve_paths(base_dir: Path) -> TeamMemoryConfig
Source code in src/naqsha/memory/sharing.py 
23
24
25
26
27
def resolve_paths(self, base_dir: Path) -> TeamMemoryConfig:
    path = self.db_path
    if not path.is_absolute():
        path = (base_dir / path).resolve()
    return TeamMemoryConfig(type=self.type, db_path=path, embeddings=self.embeddings)

SimpleMemCrossMemoryPort

SQLite-backed cross-session durable memory.

Source code in src/naqsha/memory/simplemem_cross.py 
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
class SimpleMemCrossMemoryPort:
    """SQLite-backed cross-session durable memory."""

    def __init__(self, *, project: str, database_path: Path) -> None:
        self._project = project
        self._db_path = Path(database_path)
        self._db_path.parent.mkdir(parents=True, exist_ok=True)
        self._conn = sqlite3.connect(self._db_path, isolation_level=None)
        self._conn.row_factory = sqlite3.Row
        self._initialize_schema()
        self._run_id: str | None = None
        self._query: str | None = None
        self._pending: list[tuple[str, ToolObservation]] = []

    def close(self) -> None:
        self._conn.close()

    # --- internals ---------------------------------------------------------

    def _initialize_schema(self) -> None:
        self._conn.executescript(
            """
            PRAGMA journal_mode=WAL;
            CREATE TABLE IF NOT EXISTS memory_entries (
                project TEXT NOT NULL,
                created_ts REAL NOT NULL,
                content TEXT NOT NULL,
                provenance TEXT NOT NULL
            );
            CREATE INDEX IF NOT EXISTS idx_memory_project_ts
              ON memory_entries(project, created_ts DESC);
            """
        )

    def _select_ranked_rows(self, query: str, limit: int) -> list[_RankedRow]:
        tokens = _tokenize(query)
        rows = self._conn.execute(
            """
            SELECT created_ts, content, provenance
              FROM memory_entries
             WHERE project = ?
             ORDER BY created_ts DESC
             LIMIT ?
            """,
            (self._project, limit),
        ).fetchall()
        ranked: list[_RankedRow] = []
        for row in rows:
            created_ts = float(row["created_ts"])
            content_s = row["content"]
            prov_s = row["provenance"]
            if tokens:
                hay = content_s.lower()
                hits = sum(1 for tok in tokens if _haystack_matches_token(hay, tok))
                key = (hits, created_ts)
            else:
                key = (0, created_ts)
            ranked.append(
                _RankedRow(
                    created_ts=created_ts,
                    content=content_s,
                    provenance=prov_s,
                    rank=key,
                )
            )
        ranked.sort(key=lambda r: r.rank, reverse=True)
        return ranked

    # --- Memory Port -------------------------------------------------------

    def start_run(self, run_id: str, query: str) -> None:
        self._run_id = run_id
        self._query = query
        self._pending.clear()

    def retrieve(self, query: str, token_budget: int) -> list[MemoryRecord]:
        # Session query seeds relevance for this turn; callers may diverge retrieve text.
        if query.strip():
            self._query = query
        chars_left = _approx_chars_for_tokens(token_budget)
        if chars_left == 0:
            return []

        qtext = query.strip() or (self._query or "")
        candidates = self._select_ranked_rows(qtext, limit=512)
        out: list[MemoryRecord] = []
        seen_provenance: set[str] = set()
        tokens = _tokenize(qtext)

        for row in candidates:
            key = row.provenance
            if key in seen_provenance:
                continue
            if tokens:
                lowered = row.content.lower()
                if (
                    sum(1 for tok in tokens if _haystack_matches_token(lowered, tok))
                    == 0
                ):
                    continue
            seen_provenance.add(key)
            block = _wrap_memory_evidence(row.provenance, row.content)
            provenance_note = (
                f"created_ts={row.created_ts:.3f};evidence={row.provenance};project={self._project}"
            )
            if len(block) > chars_left:
                if out:
                    break
                # First block must still respect the budget; trim body only (keep delimiters).
                overhead = len(block) - len(row.content)
                body_budget = max(0, chars_left - overhead)
                trimmed = row.content.strip()[:body_budget].rstrip()
                block = _wrap_memory_evidence(row.provenance, trimmed)
                if len(block) > chars_left:
                    continue
            out.append(MemoryRecord(content=block, provenance=provenance_note))
            chars_left -= len(block)
            if chars_left <= 0:
                break
        return out

    def record_observation(self, run_id: str, tool: str, observation: ToolObservation) -> None:
        if run_id != self._run_id:
            raise ValueError(
                f"Observation recorded for mismatched run_id {run_id!r}; "
                f"expected active {self._run_id!r}."
            )
        if observation.ok:
            self._pending.append((tool, observation))

    def finish_run(self, run_id: str, answer: str | None) -> None:
        if self._run_id != run_id:
            # Best-effort: runtime finally may still fire after a partially failed setup.
            return
        ts = time.time()
        q = self._query or ""

        cur = self._conn.cursor()
        cur.execute("BEGIN")
        try:
            for idx, (tool_name, observation) in enumerate(self._pending):
                line = observation.content.strip()
                if not line:
                    continue
                ts_row = ts + float(idx + 1) * 1e-9
                prov = f"{run_id}:{tool_name}"
                cur.execute(
                    """
                    INSERT INTO memory_entries (project, created_ts, content, provenance)
                    VALUES (?, ?, ?, ?)
                    """,
                    (self._project, ts_row, line, prov),
                )

            if answer and answer.strip():
                summary_ts = ts + float(len(self._pending) + 1) * 1e-9
                cur.execute(
                    """
                    INSERT INTO memory_entries (project, created_ts, content, provenance)
                    VALUES (?, ?, ?, ?)
                    """,
                    (
                        self._project,
                        summary_ts,
                        f"Query: {q}\nFinal answer: {answer.strip()}",
                        f"{run_id}:answer",
                    ),
                )
            cur.execute("COMMIT")
        except Exception:
            cur.execute("ROLLBACK")
            raise
        else:
            self._pending.clear()
            self._run_id = None
            self._query = None

close 

close() -> None
Source code in src/naqsha/memory/simplemem_cross.py 
88
89
def close(self) -> None:
    self._conn.close()

start_run 

start_run(run_id: str, query: str) -> None
Source code in src/naqsha/memory/simplemem_cross.py 
144
145
146
147
def start_run(self, run_id: str, query: str) -> None:
    self._run_id = run_id
    self._query = query
    self._pending.clear()

retrieve 

retrieve(
    query: str, token_budget: int
) -> list[MemoryRecord]
Source code in src/naqsha/memory/simplemem_cross.py 
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
def retrieve(self, query: str, token_budget: int) -> list[MemoryRecord]:
    # Session query seeds relevance for this turn; callers may diverge retrieve text.
    if query.strip():
        self._query = query
    chars_left = _approx_chars_for_tokens(token_budget)
    if chars_left == 0:
        return []

    qtext = query.strip() or (self._query or "")
    candidates = self._select_ranked_rows(qtext, limit=512)
    out: list[MemoryRecord] = []
    seen_provenance: set[str] = set()
    tokens = _tokenize(qtext)

    for row in candidates:
        key = row.provenance
        if key in seen_provenance:
            continue
        if tokens:
            lowered = row.content.lower()
            if (
                sum(1 for tok in tokens if _haystack_matches_token(lowered, tok))
                == 0
            ):
                continue
        seen_provenance.add(key)
        block = _wrap_memory_evidence(row.provenance, row.content)
        provenance_note = (
            f"created_ts={row.created_ts:.3f};evidence={row.provenance};project={self._project}"
        )
        if len(block) > chars_left:
            if out:
                break
            # First block must still respect the budget; trim body only (keep delimiters).
            overhead = len(block) - len(row.content)
            body_budget = max(0, chars_left - overhead)
            trimmed = row.content.strip()[:body_budget].rstrip()
            block = _wrap_memory_evidence(row.provenance, trimmed)
            if len(block) > chars_left:
                continue
        out.append(MemoryRecord(content=block, provenance=provenance_note))
        chars_left -= len(block)
        if chars_left <= 0:
            break
    return out

record_observation 

record_observation(
    run_id: str, tool: str, observation: ToolObservation
) -> None
Source code in src/naqsha/memory/simplemem_cross.py 
195
196
197
198
199
200
201
202
def record_observation(self, run_id: str, tool: str, observation: ToolObservation) -> None:
    if run_id != self._run_id:
        raise ValueError(
            f"Observation recorded for mismatched run_id {run_id!r}; "
            f"expected active {self._run_id!r}."
        )
    if observation.ok:
        self._pending.append((tool, observation))

finish_run 

finish_run(run_id: str, answer: str | None) -> None
Source code in src/naqsha/memory/simplemem_cross.py 
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
def finish_run(self, run_id: str, answer: str | None) -> None:
    if self._run_id != run_id:
        # Best-effort: runtime finally may still fire after a partially failed setup.
        return
    ts = time.time()
    q = self._query or ""

    cur = self._conn.cursor()
    cur.execute("BEGIN")
    try:
        for idx, (tool_name, observation) in enumerate(self._pending):
            line = observation.content.strip()
            if not line:
                continue
            ts_row = ts + float(idx + 1) * 1e-9
            prov = f"{run_id}:{tool_name}"
            cur.execute(
                """
                INSERT INTO memory_entries (project, created_ts, content, provenance)
                VALUES (?, ?, ?, ?)
                """,
                (self._project, ts_row, line, prov),
            )

        if answer and answer.strip():
            summary_ts = ts + float(len(self._pending) + 1) * 1e-9
            cur.execute(
                """
                INSERT INTO memory_entries (project, created_ts, content, provenance)
                VALUES (?, ?, ?, ?)
                """,
                (
                    self._project,
                    summary_ts,
                    f"Query: {q}\nFinal answer: {answer.strip()}",
                    f"{run_id}:answer",
                ),
            )
        cur.execute("COMMIT")
    except Exception:
        cur.execute("ROLLBACK")
        raise
    else:
        self._pending.clear()
        self._run_id = None
        self._query = None

is_ddl_statement 

is_ddl_statement(sql: str) -> bool

Check if SQL statement appears to be a DDL statement.

Parameters:

Name Type Description Default
sql str

SQL statement to check

 required

Returns:

Type Description
bool

True if the statement looks like DDL (CREATE, ALTER, DROP, etc.)
Source code in src/naqsha/memory/ddl.py 
72
73
74
75
76
77
78
79
80
81
82
83
def is_ddl_statement(sql: str) -> bool:
    """Check if SQL statement appears to be a DDL statement.

    Args:
        sql: SQL statement to check

    Returns:
        True if the statement looks like DDL (CREATE, ALTER, DROP, etc.)
    """
    sql_stripped = sql.strip().upper()
    ddl_keywords = ["CREATE", "ALTER", "DROP", "TRUNCATE"]
    return any(sql_stripped.startswith(keyword) for keyword in ddl_keywords)

validate_ddl 

validate_ddl(sql: str) -> None

Validate that SQL statement is an allowed DDL operation.

Parameters:

Name Type Description Default
sql str

SQL statement to validate

 required

Raises:

Type Description
ForbiddenDDLError

If the statement is not in the safelist
Source code in src/naqsha/memory/ddl.py 
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
def validate_ddl(sql: str) -> None:
    """Validate that SQL statement is an allowed DDL operation.

    Args:
        sql: SQL statement to validate

    Raises:
        ForbiddenDDLError: If the statement is not in the safelist
    """
    sql_stripped = sql.strip()

    if not sql_stripped:
        raise ForbiddenDDLError("Empty SQL statement")

    # Check for forbidden keywords first
    sql_upper = sql_stripped.upper()
    for keyword in _FORBIDDEN_KEYWORDS:
        if keyword in sql_upper:
            raise ForbiddenDDLError(
                f"Forbidden DDL keyword '{keyword}' detected. "
                f"Only CREATE TABLE, CREATE INDEX, and ALTER TABLE ADD COLUMN are permitted."
            )

    # Check if it matches any allowed pattern
    for pattern in _ALLOWED_DDL_PATTERNS:
        if pattern.match(sql_stripped):
            return

    # If we get here, it's not an allowed DDL statement
    raise ForbiddenDDLError(
        f"DDL statement not in safelist. "
        f"Only CREATE TABLE, CREATE INDEX, and ALTER TABLE ADD COLUMN are permitted. "
        f"Got: {sql_stripped[:100]}"
    )

open_team_memory_engine 

open_team_memory_engine(
    workspace_root: Path, config: TeamMemoryConfig
) -> DynamicMemoryEngine

Open the team SQLite database under the workspace root.

The same engine instance must be passed to every agent CoreRuntime in that team so shared tables are truly shared.

Source code in src/naqsha/memory/sharing.py 
30
31
32
33
34
35
36
37
38
39
40
41
42
def open_team_memory_engine(workspace_root: Path, config: TeamMemoryConfig) -> DynamicMemoryEngine:
    """Open the team SQLite database under the workspace root.

    The same engine instance must be passed to every agent CoreRuntime in that team so
    shared tables are truly shared.
    """
    resolved = config.resolve_paths(workspace_root)
    if resolved.type != "sqlite":
        raise ValueError(f"Unsupported memory type {resolved.type!r}; expected 'sqlite'.")
    return DynamicMemoryEngine(
        db_path=resolved.db_path,
        enable_embeddings=resolved.embeddings,
    )