feat: implement daemon model with registry and single-asset assignment (BAT-010, REQ-023/028)

Core Implementation (BAT-010):
- Add daemon registry schema and CRUD operations (req-023)
- Implement single-daemon asset assignment constraint (one asset per daemon)
- Add daemon_id field to service_token for daemon identification
- Add daemon.py CLI module for daemon management commands
- Add 20260513_0016_daemon_model.py migration for daemon tables

API Changes:
- GET /v1/assets: include daemon_id in AssetResponse
- POST /v1/assets/{asset_id}/ledger/trades: require daemon token binding
- GET /v1/user_info: include daemon metadata
- Enhanced service-token rotate to preserve daemon bindings

CLI Commands:
- pfhub daemon register --name <name> --runtime-id <runtime-id>
- pfhub daemon list
- pfhub daemon get <daemon-id>
- pfhub daemon revoke <daemon-id>

Requirements (REQ-023, REQ-028):
- REQ-023: Daemon Registry and Service Token Daemon Binding (Interface, Approved)
  * AC-1: daemon CRUD operations
  * AC-2: service token daemon binding for POST /v1/assets/{asset_id}/ledger/trades
  * AC-3: single daemon per asset (exact-one assignment model)
- REQ-028: README.md Daemon Provisioning Runbook update (Documentation, Approved)

Documentation:
- Add dcy-006 Discovery for daemon-id and single-daemon asset assignment
- Update README.md with Daemon Provisioning Runbook section
- Add BAT-010 batch documentation with planning, implementation, verification
- Update rel-006 release document (daemon model implementation)
- Update data-model.md with daemon and daemon_asset_assignment tables
- Update interface-contract.md with daemon registry APIs
- Update runtime-flows.md with daemon lifecycle flows

Breaking Changes:
- REQ-005~009, REQ-011~012 status reverted to Approved due to revised contract
  * daemon_id field addition and exact-one asset assignment requirement
  * Existing Implemented evidence invalidated by contract revision
  * BAT-001/002/008 re-verification required

Testing:
- Add test_daemon_registry.py for daemon CRUD operations
- Add test_daemon_assignment.py for single-asset assignment constraints
- Update test_auth.py, test_cli.py, test_v1.py for daemon support
- Verify daemon token binding in trade endpoint authorization

Impact:
- Database: 2 new tables (daemons, daemon_asset_assignments)
- Migration: 20260513_0016 creates daemon model schema
- CLI: 4 new daemon subcommands
- API: daemon_id propagation through v1 endpoints
- Verification: BAT-001/002/008 re-verification gates next release
This commit is contained in:
2026-05-13 23:07:24 +00:00
parent f0a6caa178
commit e6b6a8715b
45 changed files with 2688 additions and 167 deletions
+234 -8
View File
@@ -11,7 +11,7 @@ from typing import Any
import psycopg
from psycopg import sql
from app.domain.models import AuthToken, LedgerEntry, Portfolio, PortfolioComment, PortfolioTag, StrategySet, TokenPrincipal, User
from app.domain.models import AuthToken, Daemon, LedgerEntry, Portfolio, PortfolioComment, PortfolioTag, StrategySet, TokenPrincipal, User
_SELECT_USER = """
SELECT
@@ -39,7 +39,8 @@ _SELECT_AUTH_TOKEN = """
last_used_at,
revoked_at,
created_at,
runtime_id
runtime_id,
daemon_id
FROM auth_tokens
"""
@@ -59,7 +60,8 @@ _SELECT_PORTFOLIO = """
owner_user_id,
created_at,
updated_at,
last_trade_at
last_trade_at,
managed_by_daemon_id
FROM portfolios
"""
@@ -143,6 +145,7 @@ class PostgresPortfolioRepository:
"""
)
cur.execute("ALTER TABLE auth_tokens ADD COLUMN IF NOT EXISTS runtime_id TEXT")
cur.execute("ALTER TABLE auth_tokens ADD COLUMN IF NOT EXISTS daemon_id TEXT")
cur.execute(
"""
CREATE INDEX IF NOT EXISTS idx_auth_tokens_user_created
@@ -197,6 +200,7 @@ class PostgresPortfolioRepository:
)
cur.execute("ALTER TABLE portfolios ADD COLUMN IF NOT EXISTS name TEXT")
cur.execute("ALTER TABLE portfolios ADD COLUMN IF NOT EXISTS owner_user_id TEXT")
cur.execute("ALTER TABLE portfolios ADD COLUMN IF NOT EXISTS managed_by_daemon_id TEXT")
cur.execute("UPDATE portfolios SET name = symbol WHERE name IS NULL OR name = ''")
cur.execute("ALTER TABLE portfolios ALTER COLUMN name SET NOT NULL")
cur.execute(
@@ -263,6 +267,26 @@ class PostgresPortfolioRepository:
ON portfolio_tags (tag_value, portfolio_id)
"""
)
cur.execute(
"""
CREATE TABLE IF NOT EXISTS daemons (
id TEXT PRIMARY KEY,
owner_user_id TEXT NOT NULL,
daemon_slug TEXT NOT NULL,
display_name TEXT,
is_active BOOLEAN NOT NULL DEFAULT TRUE,
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
UNIQUE (owner_user_id, daemon_slug)
)
"""
)
cur.execute(
"""
CREATE INDEX IF NOT EXISTS idx_daemons_owner
ON daemons (owner_user_id)
"""
)
cur.execute("SELECT EXISTS (SELECT 1 FROM pg_extension WHERE extname = 'timescaledb')")
if cur.fetchone()[0]:
cur.execute(
@@ -304,6 +328,7 @@ class PostgresPortfolioRepository:
revoked_at=row[8] if len(row) > 8 else None,
created_at=row[9] if len(row) > 9 else None,
runtime_id=row[10] if len(row) > 10 else None,
daemon_id=row[11] if len(row) > 11 else None,
)
@staticmethod
@@ -336,6 +361,7 @@ class PostgresPortfolioRepository:
created_at=row[12] if len(row) > 12 else None,
updated_at=row[13] if len(row) > 13 else None,
last_trade_at=row[14] if len(row) > 14 else None,
managed_by_daemon_id=row[15] if len(row) > 15 else None,
)
@staticmethod
@@ -537,6 +563,7 @@ class PostgresPortfolioRepository:
token_type: str,
expires_at: datetime | None = None,
runtime_id: str | None = None,
daemon_id: str | None = None,
) -> AuthToken:
self.ensure_schema()
with self._connect().cursor() as cur:
@@ -547,14 +574,16 @@ class PostgresPortfolioRepository:
cur.execute(
"""
INSERT INTO auth_tokens (
id, user_id, name, token_prefix, token_hash, permission, token_type, expires_at, runtime_id
id, user_id, name, token_prefix, token_hash, permission, token_type,
expires_at, runtime_id, daemon_id
)
VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s)
VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
RETURNING
id, user_id, name, token_prefix, permission, token_type,
expires_at, last_used_at, revoked_at, created_at, runtime_id
expires_at, last_used_at, revoked_at, created_at, runtime_id, daemon_id
""",
(token_id, user_id, name, token_prefix, token_hash, permission, token_type, expires_at, runtime_id),
(token_id, user_id, name, token_prefix, token_hash, permission, token_type,
expires_at, runtime_id, daemon_id),
)
row = cur.fetchone()
return self._row_to_auth_token(row)
@@ -617,7 +646,7 @@ class PostgresPortfolioRepository:
with self._connect().cursor() as cur:
cur.execute(
"""
SELECT t.id, t.user_id, t.permission, t.token_type, t.runtime_id
SELECT t.id, t.user_id, t.permission, t.token_type, t.runtime_id, t.daemon_id
FROM auth_tokens t
WHERE token_hash = %s
AND t.revoked_at IS NULL
@@ -635,6 +664,7 @@ class PostgresPortfolioRepository:
permission=row[2],
token_type=row[3],
runtime_id=row[4] if len(row) > 4 else None,
daemon_id=row[5] if len(row) > 5 else None,
)
def get(self, portfolio_id: str) -> Portfolio:
@@ -794,6 +824,7 @@ class PostgresPortfolioRepository:
include_tags: list[str] | None = None,
exclude_tags: list[str] | None = None,
match_mode: str | None = None,
managed_by_daemon_id: str | None = None,
) -> list[Portfolio]:
self.ensure_schema()
query = _SELECT_PORTFOLIO
@@ -808,6 +839,9 @@ class PostgresPortfolioRepository:
if owner_user_id is not None:
clauses.append("owner_user_id = %s")
params.append(owner_user_id)
if managed_by_daemon_id is not None:
clauses.append("managed_by_daemon_id = %s")
params.append(managed_by_daemon_id)
normalized_mode = (match_mode or "OR").upper()
if include_tags:
if normalized_mode == "AND":
@@ -1302,6 +1336,198 @@ class PostgresPortfolioRepository:
if cur.fetchone() is None:
raise ValueError("댓글을 찾을 수 없습니다")
def list_portfolios_with_strategies(
self,
owner_user_id: str,
*,
include_disabled: bool = True,
managed_by_daemon_id: str | None = None,
) -> list[Portfolio]:
status_filter = None if include_disabled else "enabled"
return self.list_portfolios(
owner_user_id=owner_user_id,
status=status_filter,
managed_by_daemon_id=managed_by_daemon_id,
)
def rotate_service_token(self, name: str) -> tuple[str, AuthToken]:
from app.core.security import generate_token, hash_token, token_prefix as _token_prefix
self.ensure_schema()
with self._connect().cursor() as cur:
cur.execute(
_SELECT_AUTH_TOKEN + "WHERE name = %s AND token_type = 'service' AND revoked_at IS NULL",
(name,),
)
existing_row = cur.fetchone()
if existing_row is None:
raise ValueError(f"서비스 토큰을 찾을 수 없습니다: {name}")
existing = self._row_to_auth_token(existing_row)
new_raw = generate_token()
new_runtime_id = str(uuid.uuid4())
new_hash = hash_token(new_raw)
new_prefix = _token_prefix(new_raw)
cur.execute("UPDATE auth_tokens SET revoked_at = NOW() WHERE id = %s", (existing.id,))
token_id = str(uuid.uuid4())
cur.execute(
"""
INSERT INTO auth_tokens (
id, user_id, name, token_prefix, token_hash, permission, token_type,
expires_at, runtime_id, daemon_id
)
VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s)
RETURNING
id, user_id, name, token_prefix, permission, token_type,
expires_at, last_used_at, revoked_at, created_at, runtime_id, daemon_id
""",
(
token_id,
existing.user_id,
name,
new_prefix,
new_hash,
existing.permission,
"service",
existing.expires_at,
new_runtime_id,
existing.daemon_id,
),
)
row = cur.fetchone()
return new_raw, self._row_to_auth_token(row)
@staticmethod
def _row_to_daemon(row: Any) -> Daemon:
if row is None:
raise ValueError("데몬을 찾을 수 없습니다")
return Daemon(
id=row[0],
owner_user_id=row[1],
daemon_slug=row[2],
display_name=row[3],
is_active=bool(row[4]),
created_at=row[5] if len(row) > 5 else None,
updated_at=row[6] if len(row) > 6 else None,
)
def create_daemon(
self,
owner_user_id: str,
*,
daemon_slug: str,
display_name: str | None = None,
) -> Daemon:
self.ensure_schema()
with self._connect().cursor() as cur:
cur.execute("SELECT 1 FROM users WHERE id = %s", (owner_user_id,))
if cur.fetchone() is None:
raise ValueError("사용자를 찾을 수 없습니다")
daemon_id = str(uuid.uuid4())
try:
cur.execute(
"""
INSERT INTO daemons (id, owner_user_id, daemon_slug, display_name)
VALUES (%s, %s, %s, %s)
RETURNING id, owner_user_id, daemon_slug, display_name, is_active, created_at, updated_at
""",
(daemon_id, owner_user_id, daemon_slug, display_name),
)
except psycopg.errors.UniqueViolation as exc:
raise ValueError(f"이미 존재하는 daemon slug입니다: {daemon_slug}") from exc
row = cur.fetchone()
return self._row_to_daemon(row)
def get_daemon_by_id(self, daemon_id: str) -> Daemon | None:
self.ensure_schema()
with self._connect().cursor() as cur:
cur.execute(
"""
SELECT id, owner_user_id, daemon_slug, display_name, is_active, created_at, updated_at
FROM daemons WHERE id = %s
""",
(daemon_id,),
)
row = cur.fetchone()
if row is None:
return None
return self._row_to_daemon(row)
def get_daemon_by_slug(self, owner_user_id: str, daemon_slug: str) -> Daemon | None:
self.ensure_schema()
with self._connect().cursor() as cur:
cur.execute(
"""
SELECT id, owner_user_id, daemon_slug, display_name, is_active, created_at, updated_at
FROM daemons WHERE owner_user_id = %s AND daemon_slug = %s
""",
(owner_user_id, daemon_slug),
)
row = cur.fetchone()
if row is None:
return None
return self._row_to_daemon(row)
def list_daemons_for_user(self, owner_user_id: str) -> list[Daemon]:
self.ensure_schema()
with self._connect().cursor() as cur:
cur.execute(
"""
SELECT id, owner_user_id, daemon_slug, display_name, is_active, created_at, updated_at
FROM daemons WHERE owner_user_id = %s ORDER BY created_at ASC
""",
(owner_user_id,),
)
rows = cur.fetchall()
return [self._row_to_daemon(row) for row in rows]
def update_daemon(
self,
daemon_id: str,
*,
display_name: str | None = None,
is_active: bool | None = None,
) -> Daemon:
self.ensure_schema()
with self._connect().cursor() as cur:
cur.execute(
"""
SELECT id, owner_user_id, daemon_slug, display_name, is_active, created_at, updated_at
FROM daemons WHERE id = %s
""",
(daemon_id,),
)
existing_row = cur.fetchone()
if existing_row is None:
raise ValueError("데몬을 찾을 수 없습니다")
assignments: list[str] = []
params: list[object] = []
if display_name is not None:
assignments.append("display_name = %s")
params.append(display_name)
if is_active is not None:
assignments.append("is_active = %s")
params.append(is_active)
if not assignments:
return self._row_to_daemon(existing_row)
assignments.append("updated_at = NOW()")
cur.execute(
f"""
UPDATE daemons
SET {', '.join(assignments)}
WHERE id = %s
RETURNING id, owner_user_id, daemon_slug, display_name, is_active, created_at, updated_at
""",
(*params, daemon_id),
)
row = cur.fetchone()
return self._row_to_daemon(row)
def database_exists(config: PortfolioDBConfig, *, maintenance_db: str = "postgres") -> bool:
with psycopg.connect(