personal memory agent
pairing: add cryptography direct dep + pairing config defaults
+168
+1
pyproject.toml
+1
pyproject.toml
+90
tests/test_pairing_config.py
+90
tests/test_pairing_config.py
···
1
+
# SPDX-License-Identifier: AGPL-3.0-only
2
+
# Copyright (c) 2026 sol pbc
3
+
4
+
from __future__ import annotations
5
+
6
+
import json
7
+
from pathlib import Path
8
+
9
+
from think.pairing import config
10
+
11
+
12
+
def _write_config(journal: Path, payload: dict) -> None:
13
+
config_path = journal / "config" / "journal.json"
14
+
config_path.write_text(json.dumps(payload), encoding="utf-8")
15
+
16
+
17
+
def _read_config(journal: Path) -> dict:
18
+
return json.loads((journal / "config" / "journal.json").read_text(encoding="utf-8"))
19
+
20
+
21
+
def test_pairing_config_defaults(journal_copy):
22
+
payload = _read_config(journal_copy)
23
+
payload.pop("pairing", None)
24
+
payload["identity"] = {"name": "", "preferred": ""}
25
+
_write_config(journal_copy, payload)
26
+
27
+
assert config.get_host_url() == "http://localhost:5015"
28
+
assert config.get_token_ttl_seconds() == 600
29
+
assert config.get_owner_identity() == ""
30
+
31
+
32
+
def test_pairing_host_url_reads_trimmed_value(journal_copy):
33
+
payload = _read_config(journal_copy)
34
+
payload["pairing"] = {
35
+
"host_url": " https://example.test/base ",
36
+
"token_ttl_seconds": 600,
37
+
}
38
+
_write_config(journal_copy, payload)
39
+
40
+
assert config.get_host_url() == "https://example.test/base"
41
+
42
+
43
+
def test_pairing_host_url_synthesizes_recorded_convey_port(journal_copy):
44
+
payload = _read_config(journal_copy)
45
+
payload["pairing"] = {"host_url": None}
46
+
_write_config(journal_copy, payload)
47
+
health_dir = journal_copy / "health"
48
+
health_dir.mkdir(parents=True, exist_ok=True)
49
+
(health_dir / "convey.port").write_text("6123", encoding="utf-8")
50
+
51
+
assert config.get_host_url() == "http://localhost:6123"
52
+
53
+
54
+
def test_pairing_token_ttl_clamps(journal_copy):
55
+
payload = _read_config(journal_copy)
56
+
57
+
payload["pairing"] = {"token_ttl_seconds": 59}
58
+
_write_config(journal_copy, payload)
59
+
assert config.get_token_ttl_seconds() == 60
60
+
61
+
payload["pairing"] = {"token_ttl_seconds": 60}
62
+
_write_config(journal_copy, payload)
63
+
assert config.get_token_ttl_seconds() == 60
64
+
65
+
payload["pairing"] = {"token_ttl_seconds": 600}
66
+
_write_config(journal_copy, payload)
67
+
assert config.get_token_ttl_seconds() == 600
68
+
69
+
payload["pairing"] = {"token_ttl_seconds": 3600}
70
+
_write_config(journal_copy, payload)
71
+
assert config.get_token_ttl_seconds() == 3600
72
+
73
+
payload["pairing"] = {"token_ttl_seconds": 3601}
74
+
_write_config(journal_copy, payload)
75
+
assert config.get_token_ttl_seconds() == 3600
76
+
77
+
78
+
def test_pairing_owner_identity_fallbacks(journal_copy):
79
+
payload = _read_config(journal_copy)
80
+
payload["identity"] = {"name": "Sol", "preferred": "Sol Preferred"}
81
+
_write_config(journal_copy, payload)
82
+
assert config.get_owner_identity() == "Sol Preferred"
83
+
84
+
payload["identity"] = {"name": "Sol", "preferred": " "}
85
+
_write_config(journal_copy, payload)
86
+
assert config.get_owner_identity() == "Sol"
87
+
88
+
payload["identity"] = {"name": " ", "preferred": None}
89
+
_write_config(journal_copy, payload)
90
+
assert config.get_owner_identity() == ""
+4
think/journal_default.json
+4
think/journal_default.json
+4
think/pairing/__init__.py
+4
think/pairing/__init__.py
+67
think/pairing/config.py
+67
think/pairing/config.py
···
1
+
# SPDX-License-Identifier: AGPL-3.0-only
2
+
# Copyright (c) 2026 sol pbc
3
+
4
+
"""Pairing config readers."""
5
+
6
+
from __future__ import annotations
7
+
8
+
from typing import Any
9
+
10
+
from think.service import DEFAULT_SERVICE_PORT
11
+
from think.utils import get_config, read_service_port
12
+
13
+
DEFAULT_TOKEN_TTL_SECONDS = 600
14
+
MIN_TOKEN_TTL_SECONDS = 60
15
+
MAX_TOKEN_TTL_SECONDS = 3600
16
+
17
+
18
+
def _pairing_config() -> dict[str, Any]:
19
+
config = get_config()
20
+
pairing = config.get("pairing")
21
+
return pairing if isinstance(pairing, dict) else {}
22
+
23
+
24
+
def _clean_str(value: Any) -> str | None:
25
+
if not isinstance(value, str):
26
+
return None
27
+
cleaned = value.strip()
28
+
return cleaned or None
29
+
30
+
31
+
def get_host_url() -> str:
32
+
configured = _clean_str(_pairing_config().get("host_url"))
33
+
if configured is not None:
34
+
return configured
35
+
convey_port = read_service_port("convey") or DEFAULT_SERVICE_PORT
36
+
return f"http://localhost:{convey_port}"
37
+
38
+
39
+
def get_token_ttl_seconds() -> int:
40
+
configured = _pairing_config().get("token_ttl_seconds")
41
+
try:
42
+
ttl_seconds = int(configured)
43
+
except (TypeError, ValueError):
44
+
ttl_seconds = DEFAULT_TOKEN_TTL_SECONDS
45
+
return max(MIN_TOKEN_TTL_SECONDS, min(MAX_TOKEN_TTL_SECONDS, ttl_seconds))
46
+
47
+
48
+
def get_owner_identity() -> str:
49
+
config = get_config()
50
+
identity = config.get("identity")
51
+
if not isinstance(identity, dict):
52
+
return ""
53
+
preferred = _clean_str(identity.get("preferred"))
54
+
if preferred is not None:
55
+
return preferred
56
+
name = _clean_str(identity.get("name"))
57
+
return name or ""
58
+
59
+
60
+
__all__ = [
61
+
"DEFAULT_TOKEN_TTL_SECONDS",
62
+
"MAX_TOKEN_TTL_SECONDS",
63
+
"MIN_TOKEN_TTL_SECONDS",
64
+
"get_host_url",
65
+
"get_owner_identity",
66
+
"get_token_ttl_seconds",
67
+
]
+2
uv.lock
+2
uv.lock
···
3562
3562
{ name = "anthropic" },
3563
3563
{ name = "av" },
3564
3564
{ name = "blessed" },
3565
+
{ name = "cryptography" },
3565
3566
{ name = "desktop-notifier" },
3566
3567
{ name = "faster-whisper" },
3567
3568
{ name = "flask", extra = ["async"] },
···
3614
3615
{ name = "anthropic" },
3615
3616
{ name = "av" },
3616
3617
{ name = "blessed", specifier = ">=1.20.0" },
3618
+
{ name = "cryptography", specifier = ">=42" },
3617
3619
{ name = "desktop-notifier" },
3618
3620
{ name = "faster-whisper", specifier = ">=1.0.0" },
3619
3621
{ name = "flask", extras = ["async"] },