Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
42 changes: 32 additions & 10 deletions bench/ecommerce_demo.py
Original file line number Diff line number Diff line change
Expand Up @@ -28,7 +28,13 @@
from lang2sql.harness.loop import agent_loop
from lang2sql.safety.pipeline import SafetyPipeline
from lang2sql.tenancy.concierge import ContextConcierge
from lang2sql.tools.semantic_federation import FedEntry, _kv_key, _render_effective, _load_all, _resolve_term
from lang2sql.tools.semantic_federation import (
FedEntry,
_kv_key,
_render_effective,
_load_all,
_resolve_term,
)

# Stable IDs for the demo guild and its two channels.
GUILD = "acme-shop"
Expand All @@ -50,7 +56,9 @@ def _finance_identity() -> Identity:
return Identity(user_id="evan", guild_id=GUILD, channel_id=CH_FINANCE)


def _define_term(store: SqliteStore, scope: str, term: str, layer: str, entity: str, definition: str) -> None:
def _define_term(
store: SqliteStore, scope: str, term: str, layer: str, entity: str, definition: str
) -> None:
entry = FedEntry(term=term, layer=layer, entity=entity, definition=definition)
store.kv_set(scope, _kv_key(term, layer, entity), entry.to_json())

Expand Down Expand Up @@ -93,7 +101,9 @@ async def section_1_define_metrics(store: SqliteStore) -> None:

rendered = _render_effective(store, scope, channel_id, ident.user_id)
lines = [l for l in rendered.splitlines() if l.startswith("-")]
print(f"\nEffective layer for #{CH_MARKETING} now holds {len(lines)} definition(s):")
print(
f"\nEffective layer for #{CH_MARKETING} now holds {len(lines)} definition(s):"
)
print(rendered)


Expand All @@ -104,10 +114,22 @@ async def section_2_federation(store: SqliteStore) -> None:
mkt = _marketing_identity()
fin = _finance_identity()

_define_term(store, GUILD, "active_user", "channel", CH_MARKETING,
"user with a login event in the last 30 days")
_define_term(store, GUILD, "active_user", "channel", CH_FINANCE,
"user with an active paid subscription")
_define_term(
store,
GUILD,
"active_user",
"channel",
CH_MARKETING,
"user with a login event in the last 30 days",
)
_define_term(
store,
GUILD,
"active_user",
"channel",
CH_FINANCE,
"user with an active paid subscription",
)

print("Defined 'active_user' independently in two channels.\n")
print("Now resolving the *effective* definition each channel sees")
Expand All @@ -127,9 +149,9 @@ async def section_2_federation(store: SqliteStore) -> None:
print(f" #{CH_MARKETING:<10} active_user → {mkt_def}")
print(f" #{CH_FINANCE:<10} active_user → {fin_def}")

assert mkt_def and fin_def and mkt_def != fin_def, (
f"Federation failed: mkt_def={mkt_def!r}, fin_def={fin_def!r}"
)
assert (
mkt_def and fin_def and mkt_def != fin_def
), f"Federation failed: mkt_def={mkt_def!r}, fin_def={fin_def!r}"
print("\n ✅ Same term, two live definitions, zero conflict.")
print(" Each channel is its own branch in the federation tree;")
print(" neither overwrote the other. (Wren's single MDL cannot do this.)")
Expand Down
1 change: 1 addition & 0 deletions pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -15,6 +15,7 @@ dependencies = [
"discord.py>=2.3,<3.0", # Phase 1 frontend transport
"cryptography>=42.0", # EncryptedSecrets at-rest encryption
"sqlalchemy>=2.0", # generic DB explorer (one adapter, many engines)
"PyYAML>=6.0", # OKF bundle frontmatter serialization
]

[project.optional-dependencies]
Expand Down
12 changes: 9 additions & 3 deletions src/lang2sql/adapters/db/d1_explorer.py
Original file line number Diff line number Diff line change
Expand Up @@ -43,7 +43,9 @@ def __init__(
) -> None:
self.account_id = account_id
self.database_id = database_id
self._token = token if token is not None else os.environ.get("CLOUDFLARE_API_TOKEN")
self._token = (
token if token is not None else os.environ.get("CLOUDFLARE_API_TOKEN")
)
self._timeout = timeout
self._transport = transport or self._http_transport

Expand All @@ -59,7 +61,9 @@ async def list_tables(self) -> list[Table]:
async def describe_table(self, name: str) -> Table:
rows = await self._query(f"PRAGMA table_info({_ident(name)})")
cols = [
Column(name=r["name"], type=r["type"] or "", nullable=not bool(r["notnull"]))
Column(
name=r["name"], type=r["type"] or "", nullable=not bool(r["notnull"])
)
for r in rows
]
return Table(name=name, schema="", columns=cols)
Expand All @@ -86,7 +90,9 @@ async def _query(self, sql: str, params: list | None = None) -> list[dict]:

def _http_transport(self, sql: str, params: list) -> dict:
if not self._token:
raise RuntimeError("CLOUDFLARE_API_TOKEN not set (D1 requires an API token)")
raise RuntimeError(
"CLOUDFLARE_API_TOKEN not set (D1 requires an API token)"
)
url = (
f"{_API_ROOT}/accounts/{self.account_id}"
f"/d1/database/{self.database_id}/query"
Expand Down
12 changes: 9 additions & 3 deletions src/lang2sql/adapters/db/dsn_builder.py
Original file line number Diff line number Diff line change
Expand Up @@ -36,7 +36,9 @@ def _quote(s: str) -> str:
return quote_plus(s, safe="")


def build_postgresql(*, host: str, port: str, database: str, user: str, password: str) -> ConnectionSpec:
def build_postgresql(
*, host: str, port: str, database: str, user: str, password: str
) -> ConnectionSpec:
# User may paste a full URL (e.g. "host/db?sslmode=require") into the host field.
# Extract just the hostname to avoid corrupting the assembled DSN.
parsed = urlsplit("//" + host)
Expand All @@ -47,7 +49,9 @@ def build_postgresql(*, host: str, port: str, database: str, user: str, password
return ConnectionSpec(dsn=dsn, extras={})


def build_mysql(*, host: str, port: str, database: str, user: str, password: str) -> ConnectionSpec:
def build_mysql(
*, host: str, port: str, database: str, user: str, password: str
) -> ConnectionSpec:
p = int(port) if port else 3306
dsn = f"mysql+pymysql://{_quote(user)}:{_quote(password)}@{host}:{p}/{database}"
return ConnectionSpec(dsn=dsn, extras={})
Expand Down Expand Up @@ -143,7 +147,9 @@ def assemble(db_type: str, fields: dict[str, str]) -> ConnectionSpec:
# Filter to the expected kwargs (modal can hand stray keys safely).
expected = {name for name, *_ in FIELD_SCHEMA[db_type]}
cleaned = {k: (v or "").strip() for k, v in fields.items() if k in expected}
missing = [n for n, _, req, _ in FIELD_SCHEMA[db_type] if req and not cleaned.get(n)]
missing = [
n for n, _, req, _ in FIELD_SCHEMA[db_type] if req and not cleaned.get(n)
]
if missing:
raise ValueError(f"missing required fields: {', '.join(missing)}")
return builder(**cleaned)
2 changes: 1 addition & 1 deletion src/lang2sql/adapters/db/factory.py
Original file line number Diff line number Diff line change
Expand Up @@ -57,7 +57,7 @@ def build_explorer(

# Normalize bare postgresql:// → postgresql+psycopg:// (psycopg3 is installed).
if scheme == "postgresql":
connection = "postgresql+psycopg" + connection[len("postgresql"):]
connection = "postgresql+psycopg" + connection[len("postgresql") :]

# Anything else is assumed to be a SQLAlchemy URL (driver loaded lazily).
return SqlAlchemyExplorer(connection, schema=schema)
Expand Down
18 changes: 15 additions & 3 deletions src/lang2sql/adapters/db/postgres_explorer.py
Original file line number Diff line number Diff line change
Expand Up @@ -18,7 +18,9 @@
columns=[
Column("id", "integer", nullable=False, description="Primary key."),
Column("amount", "numeric", nullable=False, description="Order total."),
Column("status", "text", description="pending | paid | shipped | cancelled."),
Column(
"status", "text", description="pending | paid | shipped | cancelled."
),
Column("created_at", "timestamptz", nullable=False),
],
),
Expand All @@ -36,8 +38,18 @@

_SAMPLES: dict[str, list[dict]] = {
"public.orders": [
{"id": 1, "amount": 49.90, "status": "paid", "created_at": "2026-05-01T10:00:00Z"},
{"id": 2, "amount": 12.00, "status": "pending", "created_at": "2026-05-02T14:30:00Z"},
{
"id": 1,
"amount": 49.90,
"status": "paid",
"created_at": "2026-05-01T10:00:00Z",
},
{
"id": 2,
"amount": 12.00,
"status": "pending",
"created_at": "2026-05-02T14:30:00Z",
},
],
"public.users": [
{"id": 1, "email": "alice@example.com", "created_at": "2026-04-20T08:00:00Z"},
Expand Down
4 changes: 3 additions & 1 deletion src/lang2sql/adapters/db/sqlalchemy_explorer.py
Original file line number Diff line number Diff line change
Expand Up @@ -61,7 +61,9 @@ def _list_tables_sync(self) -> list[Table]:
default = insp.default_schema_name
effective = self._schema or default
# Omit schema when it's the connection default so SQL stays unqualified.
display_schema = "" if (not self._schema or self._schema == default) else effective
display_schema = (
"" if (not self._schema or self._schema == default) else effective
)
return [
Table(name=t, schema=display_schema)
for t in insp.get_table_names(schema=self._schema)
Expand Down
4 changes: 3 additions & 1 deletion src/lang2sql/adapters/llm/fake.py
Original file line number Diff line number Diff line change
Expand Up @@ -50,7 +50,9 @@ async def complete(
)

# No tools at all → just answer.
return Completion(content="(no tools available) Hello from FakeLLM.", finish_reason="stop")
return Completion(
content="(no tools available) Hello from FakeLLM.", finish_reason="stop"
)


def _demo_args(spec: ToolSpec) -> str:
Expand Down
4 changes: 3 additions & 1 deletion src/lang2sql/adapters/llm/openai_.py
Original file line number Diff line number Diff line change
Expand Up @@ -83,7 +83,9 @@ def _post(self, payload: dict[str, Any]) -> dict[str, Any]:
try:
return json.loads(text)
except (ValueError, TypeError) as exc:
raise RuntimeError(f"OpenAI returned non-JSON response: {text[:200]!r}") from exc
raise RuntimeError(
f"OpenAI returned non-JSON response: {text[:200]!r}"
) from exc


def _strip_thinking(text: str) -> str:
Expand Down
174 changes: 174 additions & 0 deletions src/lang2sql/adapters/storage/okf_bundle.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,174 @@
"""OkfBundle — OKF(Open Knowledge Format) 기반 지식 번들 어댑터.

KV 캐시(SqliteStore)와 양방향 sync:
- export: KV → 스코프별 .md 파일 (Git 영속, 사람이 읽을 수 있는 형태)
- import_: .md 파일 → KV (번들에서 런타임 캐시 복원)

디렉토리 구조 (OKF SPEC §3):
<base_dir>/
├── guild/
│ ├── index.md
│ ├── metrics/active_user.md
│ ├── tables/orders.md
│ ├── rules/exclude_cancelled.md
│ ├── dimensions/customer_tier.md
│ └── misc/<term>.md # kind 미지정
└── channel:<ch_id>/
├── index.md
└── metrics/active_user.md

각 .md 파일 형식 (OKF SPEC §4):
---
type: Metric
title: active_user
description: "30일 내 로그인한 users"
tags: [growth, retention]
applies_to: users
synonyms: [활성화고객]
layer: guild
entity: ""
inferred: false
timestamp: 2026-07-18T...
---

(markdown body — definition 반복 또는 추가 설명)
"""

from __future__ import annotations

from datetime import datetime, timezone
from pathlib import Path
from typing import TYPE_CHECKING

import yaml

from ...tools.semantic_federation import (
FedEntry,
_KV_PREFIX,
_kv_key,
)

if TYPE_CHECKING:
from .sqlite_store import SqliteStore

_KIND_FOLDER: dict[str, str] = {
"metric": "metrics",
"table": "tables",
"rule": "rules",
"dimension": "dimensions",
}
_RESERVED = {"index.md", "log.md"}


class OkfBundle:
"""KV ↔ OKF .md 파일 양방향 sync 어댑터."""

def __init__(self, base_dir: str) -> None:
self.base_dir = Path(base_dir)

# ------------------------------------------------------------------
# Public API
# ------------------------------------------------------------------

def export(self, store: "SqliteStore", kv_scope: str) -> int:
"""KV에서 모든 FedEntry를 읽어 .md 파일로 저장. 저장된 파일 수 반환."""
raw = store.kv_list_prefix(kv_scope, _KV_PREFIX + ":")
count = 0
for _key, val in raw:
try:
entry = FedEntry.from_json(val)
except (ValueError, KeyError):
continue
path = self._concept_path(entry)
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(_entry_to_md(entry), encoding="utf-8")
count += 1
return count

def import_(self, store: "SqliteStore", kv_scope: str) -> int:
"""번들의 .md 파일을 읽어 KV로 복원. 로드된 항목 수 반환."""
count = 0
for md_file in self.base_dir.rglob("*.md"):
if md_file.name in _RESERVED:
continue
entry = _md_to_entry(md_file)
if entry is None:
continue
key = _kv_key(entry.term, entry.layer, entry.entity)
store.kv_set(kv_scope, key, entry.to_json())
count += 1
return count

# ------------------------------------------------------------------
# Helpers
# ------------------------------------------------------------------

def _scope_dir(self, entry: FedEntry) -> Path:
label = "guild" if entry.layer == "guild" else f"{entry.layer}:{entry.entity}"
return self.base_dir / label

def _concept_path(self, entry: FedEntry) -> Path:
folder = _KIND_FOLDER.get(entry.kind, "misc")
slug = entry.term.strip().lower().replace(" ", "_").replace(":", "-")
return self._scope_dir(entry) / folder / f"{slug}.md"


# ------------------------------------------------------------------
# Serialization helpers (module-level for testability)
# ------------------------------------------------------------------


def _entry_to_md(entry: FedEntry) -> str:
"""FedEntry → OKF .md 문자열 (SPEC §4.1)."""
fm: dict = {
"type": entry.kind.capitalize() if entry.kind else "Concept",
"title": entry.term,
"description": entry.definition,
}
if entry.tags:
fm["tags"] = entry.tags
if entry.applies_to:
fm["applies_to"] = entry.applies_to
if entry.synonyms:
fm["synonyms"] = entry.synonyms
fm["layer"] = entry.layer
fm["entity"] = entry.entity
fm["inferred"] = entry.inferred
fm["timestamp"] = datetime.now(timezone.utc).isoformat()

yaml_block = yaml.dump(
fm, allow_unicode=True, default_flow_style=False, sort_keys=False
)
return f"---\n{yaml_block}---\n\n{entry.definition}\n"


def _md_to_entry(path: Path) -> FedEntry | None:
"""OKF .md 파일 → FedEntry. 파싱 실패 시 None 반환."""
text = path.read_text(encoding="utf-8")
if not text.startswith("---"):
return None
try:
end = text.index("---", 3)
except ValueError:
return None
try:
fm = yaml.safe_load(text[3:end])
except yaml.YAMLError:
return None
if not isinstance(fm, dict):
return None

raw_kind = str(fm.get("type", "")).lower()
kind = raw_kind if raw_kind in _KIND_FOLDER else ""

return FedEntry(
term=str(fm.get("title", path.stem)),
layer=str(fm.get("layer", "guild")),
entity=str(fm.get("entity", "")),
definition=str(fm.get("description", "")),
synonyms=fm.get("synonyms") or [],
inferred=bool(fm.get("inferred", False)),
kind=kind,
applies_to=str(fm.get("applies_to", "")),
tags=fm.get("tags") or [],
)
Loading
Loading