personal memory agent
Add import review CLI and talents for staged item resolution
New CLI commands (sol call import) for reviewing and resolving staged
entities, facets, and config left behind by journal-to-journal import.
Three cogitate talents drive the review process through CLI commands,
with a unified wrapper that runs entity → facet → config in dependency
order. Decision logging includes resolved_by: "talent" for all
resolution entries.
+1625
+627
apps/import/call.py
+627
apps/import/call.py
···
1
+
# SPDX-License-Identifier: AGPL-3.0-only
2
+
# Copyright (c) 2026 sol pbc
3
+
4
+
"""CLI commands for import review and resolution.
5
+
6
+
Auto-discovered by ``think.call`` and mounted as ``sol call import ...``.
7
+
"""
8
+
9
+
from __future__ import annotations
10
+
11
+
import json
12
+
import os
13
+
from datetime import datetime, timezone
14
+
from importlib import import_module
15
+
from pathlib import Path
16
+
from typing import Any
17
+
18
+
import typer
19
+
20
+
from think.entities.core import EntityDict, entity_slug
21
+
from think.entities.journal import (
22
+
has_journal_principal,
23
+
journal_entity_path,
24
+
load_all_journal_entities,
25
+
save_journal_entity,
26
+
)
27
+
from think.entities.observations import load_observations, save_observations
28
+
from think.entities.relationships import (
29
+
load_facet_relationship,
30
+
save_facet_relationship,
31
+
)
32
+
from think.utils import get_journal
33
+
34
+
app = typer.Typer(help="Import review and resolution.")
35
+
36
+
ingest = import_module("apps.import.ingest")
37
+
journal_sources = import_module("apps.import.journal_sources")
38
+
39
+
_ENTITY_FILE_TYPES = {
40
+
"entity_relationship",
41
+
"entity_observations",
42
+
"detected_entities",
43
+
"activity_records",
44
+
}
45
+
_append_decision = ingest._append_decision
46
+
_categorize_field = ingest._categorize_field
47
+
_write_state_atomic = ingest._write_state_atomic
48
+
find_journal_source_by_name = journal_sources.find_journal_source_by_name
49
+
get_state_directory = journal_sources.get_state_directory
50
+
51
+
52
+
def _fail(message: str) -> None:
53
+
typer.echo(f"Error: {message}", err=True)
54
+
raise typer.Exit(1)
55
+
56
+
57
+
def _resolve_source(name: str) -> tuple[dict, str, Path]:
58
+
source = find_journal_source_by_name(name)
59
+
if not source:
60
+
_fail(
61
+
f"Import source '{name}' not found. Check available sources in "
62
+
"~/.local/share/solstone/app-storage/import/journal_sources/."
63
+
)
64
+
65
+
key = source.get("key")
66
+
if not isinstance(key, str) or len(key) < 8:
67
+
_fail(f"Import source '{name}' has an invalid key.")
68
+
69
+
key_prefix = key[:8]
70
+
state_dir = get_state_directory(key_prefix)
71
+
return source, key_prefix, state_dir
72
+
73
+
74
+
def _set_nested(cfg: dict, dotted_key: str, value: Any) -> None:
75
+
parts = dotted_key.split(".")
76
+
current = cfg
77
+
for part in parts[:-1]:
78
+
child = current.get(part)
79
+
if not isinstance(child, dict):
80
+
child = {}
81
+
current[part] = child
82
+
current = child
83
+
current[parts[-1]] = value
84
+
85
+
86
+
def _write_config(config: dict) -> None:
87
+
config_path = Path(get_journal()) / "config" / "journal.json"
88
+
config_path.parent.mkdir(parents=True, exist_ok=True)
89
+
with open(config_path, "w", encoding="utf-8") as handle:
90
+
json.dump(config, handle, indent=2, ensure_ascii=False)
91
+
handle.write("\n")
92
+
os.chmod(config_path, 0o600)
93
+
94
+
95
+
def merge_entity_fields(target: EntityDict, source: EntityDict) -> tuple[EntityDict, list[str]]:
96
+
merged: EntityDict = dict(target)
97
+
pre_merge_snapshot = dict(merged)
98
+
99
+
aka_by_lower: dict[str, str] = {}
100
+
for values in (merged.get("aka", []), source.get("aka", [])):
101
+
if not isinstance(values, list):
102
+
continue
103
+
for value in values:
104
+
if not value:
105
+
continue
106
+
key = str(value).lower()
107
+
if key not in aka_by_lower:
108
+
aka_by_lower[key] = str(value)
109
+
if aka_by_lower:
110
+
merged["aka"] = sorted(aka_by_lower.values(), key=str.lower)
111
+
112
+
merged_emails: list[str] = []
113
+
seen_emails: set[str] = set()
114
+
for values in (merged.get("emails", []), source.get("emails", [])):
115
+
if not isinstance(values, list):
116
+
continue
117
+
for value in values:
118
+
if not value:
119
+
continue
120
+
email = str(value)
121
+
key = email.lower()
122
+
if key in seen_emails:
123
+
continue
124
+
seen_emails.add(key)
125
+
merged_emails.append(email)
126
+
if merged_emails:
127
+
merged["emails"] = merged_emails
128
+
129
+
source_created = source.get("created_at")
130
+
target_created = merged.get("created_at")
131
+
if source_created is not None and target_created is not None:
132
+
merged["created_at"] = min(source_created, target_created)
133
+
elif source_created is not None:
134
+
merged["created_at"] = source_created
135
+
136
+
fields_changed = sorted(
137
+
key
138
+
for key in set(pre_merge_snapshot) | set(merged)
139
+
if pre_merge_snapshot.get(key) != merged.get(key)
140
+
)
141
+
return merged, fields_changed
142
+
143
+
144
+
def _allocate_slug(name: str) -> str | None:
145
+
base_slug = entity_slug(name)
146
+
if not base_slug:
147
+
return None
148
+
149
+
for attempt in range(1, 102):
150
+
candidate = base_slug if attempt == 1 else f"{base_slug}_{attempt}"
151
+
if not journal_entity_path(candidate).exists():
152
+
return candidate
153
+
return None
154
+
155
+
156
+
def _log_resolution(
157
+
log_path: Path,
158
+
action: str,
159
+
item_type: str,
160
+
item_id: str,
161
+
reason: str,
162
+
**extra: Any,
163
+
) -> None:
164
+
entry = {
165
+
"ts": datetime.now(timezone.utc).isoformat(),
166
+
"action": action,
167
+
"item_type": item_type,
168
+
"item_id": item_id,
169
+
"reason": reason,
170
+
"resolved_by": "talent",
171
+
}
172
+
entry.update(extra)
173
+
_append_decision(log_path, entry)
174
+
175
+
176
+
def _load_json(path: Path, default: Any) -> Any:
177
+
try:
178
+
return json.loads(path.read_text(encoding="utf-8"))
179
+
except (OSError, json.JSONDecodeError):
180
+
return default
181
+
182
+
183
+
def _load_entity_state(state_path: Path) -> dict[str, dict[str, Any]]:
184
+
entity_state = _load_json(state_path, {})
185
+
if not isinstance(entity_state, dict):
186
+
entity_state = {}
187
+
188
+
id_map = entity_state.get("id_map")
189
+
received = entity_state.get("received")
190
+
if not isinstance(id_map, dict) or not isinstance(received, dict):
191
+
return {"id_map": {}, "received": {}}
192
+
193
+
return {"id_map": dict(id_map), "received": dict(received)}
194
+
195
+
196
+
def _parse_jsonl_text(source_data: str) -> list[dict[str, Any]]:
197
+
items: list[dict[str, Any]] = []
198
+
for line_number, line in enumerate(source_data.splitlines(), start=1):
199
+
line = line.strip()
200
+
if not line:
201
+
continue
202
+
try:
203
+
item = json.loads(line)
204
+
except json.JSONDecodeError as exc:
205
+
raise ValueError(f"Invalid JSONL at line {line_number}: {exc.msg}") from exc
206
+
if not isinstance(item, dict):
207
+
raise ValueError(f"Invalid JSONL at line {line_number}: item must be an object")
208
+
items.append(item)
209
+
return items
210
+
211
+
212
+
def _append_jsonl_items(target_path: Path, items: list[dict[str, Any]]) -> None:
213
+
if not items:
214
+
return
215
+
facet_ingest = import_module("apps.import.facet_ingest")
216
+
target_path.parent.mkdir(parents=True, exist_ok=True)
217
+
with open(target_path, "ab") as handle:
218
+
handle.write(facet_ingest._serialize_jsonl(items))
219
+
220
+
221
+
def _load_config_diff(diff_path: Path) -> dict[str, dict[str, Any]]:
222
+
diff = _load_json(diff_path, {})
223
+
if not isinstance(diff, dict):
224
+
_fail("Config diff is invalid.")
225
+
return diff
226
+
227
+
228
+
def _write_config_diff(diff_path: Path, diff: dict[str, dict[str, Any]]) -> None:
229
+
diff_path.parent.mkdir(parents=True, exist_ok=True)
230
+
diff_path.write_text(
231
+
json.dumps(diff, indent=2, ensure_ascii=False) + "\n",
232
+
encoding="utf-8",
233
+
)
234
+
235
+
236
+
def _resolve_config_field(state_dir: Path, field: str, action: str) -> None:
237
+
diff_path = state_dir / "config" / "diff.json"
238
+
if not diff_path.exists():
239
+
_fail("No staged config diff found.")
240
+
241
+
diff = _load_config_diff(diff_path)
242
+
if field not in diff:
243
+
_fail(f"Config field '{field}' is not staged.")
244
+
245
+
if action not in {"apply", "keep"}:
246
+
_fail("Action must be 'apply' or 'keep'.")
247
+
248
+
diff_entry = diff[field]
249
+
if not isinstance(diff_entry, dict):
250
+
_fail(f"Config field '{field}' has invalid diff data.")
251
+
252
+
if action == "apply":
253
+
from think.utils import get_config
254
+
255
+
config = get_config()
256
+
_set_nested(config, field, diff_entry.get("source"))
257
+
_write_config(config)
258
+
log_action = "config_field_applied"
259
+
reason = "review_apply"
260
+
else:
261
+
log_action = "config_field_kept"
262
+
reason = "review_keep"
263
+
264
+
diff.pop(field)
265
+
if diff:
266
+
_write_config_diff(diff_path, diff)
267
+
else:
268
+
diff_path.unlink(missing_ok=True)
269
+
(state_dir / "config" / "source_config.json").unlink(missing_ok=True)
270
+
271
+
_log_resolution(
272
+
state_dir / "config" / "log.jsonl",
273
+
action=log_action,
274
+
item_type="config",
275
+
item_id=field,
276
+
reason=reason,
277
+
category=diff_entry.get("category", _categorize_field(field)),
278
+
source=diff_entry.get("source"),
279
+
target_previous=diff_entry.get("target"),
280
+
)
281
+
282
+
283
+
@app.command("list-staged")
284
+
def list_staged(
285
+
source: str = typer.Option(..., "--source", help="Import source name."),
286
+
area: str | None = typer.Option(None, "--area", help="Area: entities, facets, or config."),
287
+
) -> None:
288
+
_, _, state_dir = _resolve_source(source)
289
+
290
+
if area is not None and area not in {"entities", "facets", "config"}:
291
+
_fail("Area must be one of: entities, facets, config.")
292
+
293
+
if area in {None, "entities"}:
294
+
staged_dir = state_dir / "entities" / "staged"
295
+
for staged_path in sorted(staged_dir.glob("*.json")):
296
+
payload = _load_json(staged_path, {})
297
+
if not isinstance(payload, dict):
298
+
continue
299
+
line = {
300
+
"area": "entities",
301
+
"source_id": staged_path.stem,
302
+
"reason": payload.get("reason"),
303
+
"source_entity": payload.get("source_entity"),
304
+
"match_candidates": payload.get("match_candidates"),
305
+
"staged_at": payload.get("staged_at"),
306
+
}
307
+
typer.echo(json.dumps(line, ensure_ascii=False))
308
+
309
+
if area in {None, "facets"}:
310
+
staged_dir = state_dir / "facets" / "staged"
311
+
for staged_path in sorted(staged_dir.glob("**/*.staged.json")):
312
+
payload = _load_json(staged_path, {})
313
+
if not isinstance(payload, dict):
314
+
continue
315
+
relative_path = staged_path.relative_to(staged_dir)
316
+
parts = relative_path.parts
317
+
if len(parts) < 3:
318
+
continue
319
+
line = {
320
+
"area": "facets",
321
+
"staged_file": relative_path.as_posix(),
322
+
"facet": parts[0],
323
+
"file_type": parts[1],
324
+
}
325
+
line.update(payload)
326
+
typer.echo(json.dumps(line, ensure_ascii=False))
327
+
328
+
if area in {None, "config"}:
329
+
diff_path = state_dir / "config" / "diff.json"
330
+
if diff_path.exists():
331
+
diff = _load_config_diff(diff_path)
332
+
typer.echo(json.dumps({"area": "config", "diff": diff}, ensure_ascii=False))
333
+
334
+
335
+
@app.command("resolve-entity")
336
+
def resolve_entity(
337
+
source_id: str = typer.Argument(help="Source entity ID."),
338
+
action: str = typer.Argument(help="Action: merge, create, or skip."),
339
+
source: str = typer.Option(..., "--source", help="Import source name."),
340
+
target: str | None = typer.Option(None, "--target", help="Target entity ID for merge."),
341
+
) -> None:
342
+
_, _, state_dir = _resolve_source(source)
343
+
344
+
if action not in {"merge", "create", "skip"}:
345
+
_fail("Action must be 'merge', 'create', or 'skip'.")
346
+
347
+
staged_path = state_dir / "entities" / "staged" / f"{source_id}.json"
348
+
if not staged_path.exists():
349
+
_fail(f"Staged entity '{source_id}' not found.")
350
+
351
+
payload = _load_json(staged_path, {})
352
+
if not isinstance(payload, dict):
353
+
_fail(f"Staged entity '{source_id}' is invalid.")
354
+
355
+
source_entity = payload.get("source_entity")
356
+
if not isinstance(source_entity, dict):
357
+
_fail(f"Staged entity '{source_id}' is missing source_entity.")
358
+
359
+
log_path = state_dir / "entities" / "log.jsonl"
360
+
state_path = state_dir / "entities" / "state.json"
361
+
entity_state = _load_entity_state(state_path)
362
+
reason = str(payload.get("reason", ""))
363
+
match_candidates = payload.get("match_candidates")
364
+
match_tier = None
365
+
if isinstance(match_candidates, list) and match_candidates:
366
+
first_candidate = match_candidates[0]
367
+
if isinstance(first_candidate, dict):
368
+
match_tier = first_candidate.get("tier")
369
+
370
+
if action == "merge":
371
+
if not target:
372
+
_fail("--target is required for merge.")
373
+
374
+
target_entities = load_all_journal_entities()
375
+
target_entity = target_entities.get(target)
376
+
if target_entity is None:
377
+
_fail(
378
+
f"Target entity '{target}' not found. Use "
379
+
"'list-staged --source SOURCE --area entities' to check "
380
+
"match candidates, or use 'create' instead of 'merge'."
381
+
)
382
+
383
+
merged, fields_changed = merge_entity_fields(target_entity, source_entity)
384
+
save_journal_entity(merged)
385
+
entity_state["id_map"][source_id] = target
386
+
_write_state_atomic(state_path, entity_state)
387
+
staged_path.unlink()
388
+
_log_resolution(
389
+
log_path,
390
+
action="resolved_merge",
391
+
item_type="entity",
392
+
item_id=source_id,
393
+
reason=reason,
394
+
source=source_entity,
395
+
target=merged,
396
+
fields_changed=fields_changed,
397
+
match_tier=match_tier,
398
+
)
399
+
typer.echo(f"Merged {source_id} into {target}.")
400
+
return
401
+
402
+
if target is not None:
403
+
_fail("--target is only valid for merge.")
404
+
405
+
if action == "create":
406
+
created_entity = dict(source_entity)
407
+
final_id = str(created_entity.get("id") or source_id)
408
+
if reason == "id_collision" or journal_entity_path(final_id).exists():
409
+
allocated = _allocate_slug(str(created_entity.get("name", "")))
410
+
if allocated is None:
411
+
_fail(f"Unable to allocate a slug for '{created_entity.get('name', '')}'.")
412
+
final_id = allocated
413
+
created_entity["id"] = final_id
414
+
415
+
if reason == "principal_conflict" and has_journal_principal():
416
+
created_entity["is_principal"] = False
417
+
418
+
save_journal_entity(created_entity)
419
+
entity_state["id_map"][source_id] = final_id
420
+
_write_state_atomic(state_path, entity_state)
421
+
staged_path.unlink()
422
+
_log_resolution(
423
+
log_path,
424
+
action="resolved_create",
425
+
item_type="entity",
426
+
item_id=source_id,
427
+
reason=reason,
428
+
source=source_entity,
429
+
target=created_entity,
430
+
match_tier=match_tier,
431
+
fields_changed=[],
432
+
)
433
+
typer.echo(f"Created entity {final_id} from {source_id}.")
434
+
return
435
+
436
+
staged_path.unlink()
437
+
_log_resolution(
438
+
log_path,
439
+
action="resolved_skip",
440
+
item_type="entity",
441
+
item_id=source_id,
442
+
reason=reason,
443
+
source=source_entity,
444
+
target=None,
445
+
match_tier=match_tier,
446
+
fields_changed=[],
447
+
)
448
+
typer.echo(f"Skipped staged entity {source_id}.")
449
+
450
+
451
+
@app.command("resolve-facet")
452
+
def resolve_facet(
453
+
staged_file: str = typer.Argument(help="Staged file path relative to facets/staged/."),
454
+
action: str = typer.Argument(help="Action: apply or skip."),
455
+
source: str = typer.Option(..., "--source", help="Import source name."),
456
+
) -> None:
457
+
_, _, state_dir = _resolve_source(source)
458
+
459
+
if action not in {"apply", "skip"}:
460
+
_fail("Action must be 'apply' or 'skip'.")
461
+
462
+
staged_dir = state_dir / "facets" / "staged"
463
+
staged_path = staged_dir / staged_file
464
+
if not staged_path.exists():
465
+
_fail(f"Staged facet file '{staged_file}' not found.")
466
+
467
+
payload = _load_json(staged_path, {})
468
+
if not isinstance(payload, dict):
469
+
_fail(f"Staged facet file '{staged_file}' is invalid.")
470
+
471
+
parts = Path(staged_file).parts
472
+
if len(parts) < 3:
473
+
_fail(f"Staged facet file '{staged_file}' has an invalid path.")
474
+
475
+
facet_name = parts[0]
476
+
file_type = parts[1]
477
+
reason = str(payload.get("reason", ""))
478
+
log_path = state_dir / "facets" / "log.jsonl"
479
+
480
+
if reason == "facet_json_conflict":
481
+
item_id = f"{facet_name}/facet.json"
482
+
else:
483
+
item_id = f"{facet_name}/{payload.get('source_path', staged_file)}"
484
+
485
+
if action == "skip":
486
+
staged_path.unlink()
487
+
_log_resolution(
488
+
log_path,
489
+
action="resolved_skip",
490
+
item_type=file_type,
491
+
item_id=item_id,
492
+
reason=reason,
493
+
facet=facet_name,
494
+
staged_path=str(staged_path),
495
+
)
496
+
typer.echo(f"Skipped staged facet file {staged_file}.")
497
+
return
498
+
499
+
if reason == "unmapped_entity":
500
+
if file_type not in _ENTITY_FILE_TYPES:
501
+
_fail(f"Unsupported staged facet file type '{file_type}'.")
502
+
503
+
facet_ingest = import_module("apps.import.facet_ingest")
504
+
entities_state = _load_entity_state(state_dir / "entities" / "state.json")
505
+
id_map = entities_state.get("id_map", {})
506
+
source_entity_id = str(payload.get("source_entity_id", ""))
507
+
if source_entity_id not in id_map:
508
+
_fail(f"Entity {source_entity_id} has no mapping yet. Run entity review first.")
509
+
510
+
source_path = str(payload.get("source_path", ""))
511
+
source_data = str(payload.get("source_data", ""))
512
+
513
+
normalized_path, path_info = facet_ingest._parse_path(source_path, file_type)
514
+
if file_type == "entity_relationship":
515
+
parsed_data: Any = json.loads(source_data)
516
+
else:
517
+
parsed_data = _parse_jsonl_text(source_data)
518
+
519
+
remapped_data, remapped_path_info = facet_ingest._remap_entity_ids(
520
+
parsed_data, id_map, file_type, path_info
521
+
)
522
+
target_path = Path(get_journal()) / "facets" / facet_name / normalized_path
523
+
524
+
if file_type == "entity_relationship":
525
+
entity_id = remapped_path_info["entity_id"]
526
+
source_relationship = dict(remapped_data)
527
+
source_relationship["entity_id"] = entity_id
528
+
target_relationship = load_facet_relationship(facet_name, entity_id) or {}
529
+
merged_relationship = {**source_relationship, **target_relationship}
530
+
save_facet_relationship(facet_name, entity_id, merged_relationship)
531
+
elif file_type == "entity_observations":
532
+
entity_id = remapped_path_info["entity_id"]
533
+
target_observations = load_observations(facet_name, entity_id)
534
+
seen = {
535
+
(item.get("content", ""), item.get("observed_at"))
536
+
for item in target_observations
537
+
}
538
+
merged_observations = list(target_observations)
539
+
for item in remapped_data:
540
+
key = (item.get("content", ""), item.get("observed_at"))
541
+
if key in seen:
542
+
continue
543
+
seen.add(key)
544
+
merged_observations.append(item)
545
+
save_observations(facet_name, entity_id, merged_observations)
546
+
elif file_type in {"detected_entities", "activity_records"}:
547
+
existing_items = _parse_jsonl_text(target_path.read_text(encoding="utf-8")) if target_path.exists() else []
548
+
existing_ids = {item.get("id") for item in existing_items}
549
+
new_items = [item for item in remapped_data if item.get("id") not in existing_ids]
550
+
_append_jsonl_items(target_path, new_items)
551
+
else:
552
+
_fail(f"Unsupported staged facet file type '{file_type}'.")
553
+
554
+
staged_path.unlink()
555
+
_log_resolution(
556
+
log_path,
557
+
action="resolved_apply",
558
+
item_type=file_type,
559
+
item_id=item_id,
560
+
reason=reason,
561
+
facet=facet_name,
562
+
staged_path=str(staged_path),
563
+
target_path=str(target_path),
564
+
)
565
+
typer.echo(f"Applied staged facet file {staged_file}.")
566
+
return
567
+
568
+
if reason == "facet_json_conflict":
569
+
target_path = Path(get_journal()) / "facets" / facet_name / "facet.json"
570
+
target_path.parent.mkdir(parents=True, exist_ok=True)
571
+
target_path.write_text(
572
+
json.dumps(payload.get("source_content"), indent=2, ensure_ascii=False) + "\n",
573
+
encoding="utf-8",
574
+
)
575
+
staged_path.unlink()
576
+
_log_resolution(
577
+
log_path,
578
+
action="resolved_apply",
579
+
item_type=file_type,
580
+
item_id=item_id,
581
+
reason=reason,
582
+
facet=facet_name,
583
+
staged_path=str(staged_path),
584
+
target_path=str(target_path),
585
+
)
586
+
typer.echo(f"Applied staged facet file {staged_file}.")
587
+
return
588
+
589
+
_fail(f"Unsupported staged facet reason '{reason}'.")
590
+
591
+
592
+
@app.command("resolve-config")
593
+
def resolve_config(
594
+
field: str = typer.Argument(help="Dotted config field path."),
595
+
action: str = typer.Argument(help="Action: apply or keep."),
596
+
source: str = typer.Option(..., "--source", help="Import source name."),
597
+
) -> None:
598
+
_, _, state_dir = _resolve_source(source)
599
+
_resolve_config_field(state_dir, field, action)
600
+
typer.echo(f"Resolved config field {field} with action {action}.")
601
+
602
+
603
+
@app.command("resolve-config-all")
604
+
def resolve_config_all(
605
+
source: str = typer.Option(..., "--source", help="Import source name."),
606
+
category: str = typer.Option(..., "--category", help="Category: transferable or preference."),
607
+
) -> None:
608
+
_, _, state_dir = _resolve_source(source)
609
+
610
+
if category not in {"transferable", "preference"}:
611
+
_fail("Category must be 'transferable' or 'preference'.")
612
+
613
+
diff_path = state_dir / "config" / "diff.json"
614
+
if not diff_path.exists():
615
+
_fail("No staged config diff found.")
616
+
617
+
diff = _load_config_diff(diff_path)
618
+
fields = [
619
+
field
620
+
for field, diff_entry in diff.items()
621
+
if isinstance(diff_entry, dict)
622
+
and diff_entry.get("category", _categorize_field(field)) == category
623
+
]
624
+
for field in list(fields):
625
+
_resolve_config_field(state_dir, field, "apply")
626
+
627
+
typer.echo(f"Applied {len(fields)} {category} config field(s).")
+91
apps/import/talent/review.md
+91
apps/import/talent/review.md
···
1
+
{
2
+
"type": "cogitate",
3
+
"title": "Import Review",
4
+
"description": "Unified import review: resolves staged entities, then facets, then config in dependency order.",
5
+
"color": "#37474f",
6
+
"group": "Import"
7
+
}
8
+
9
+
$sol_identity
10
+
11
+
$facets
12
+
13
+
## Core Mission
14
+
15
+
Run the full import review pipeline in the correct dependency order: entities first, then facets, then config.
16
+
17
+
## Tooling
18
+
19
+
- `sol call import list-staged --source SOURCE`
20
+
- `sol call import list-staged --source SOURCE --area entities`
21
+
- `sol call import list-staged --source SOURCE --area facets`
22
+
- `sol call import list-staged --source SOURCE --area config`
23
+
- `sol call import resolve-entity SOURCE_ID merge --source SOURCE --target TARGET_ID`
24
+
- `sol call import resolve-entity SOURCE_ID create --source SOURCE`
25
+
- `sol call import resolve-entity SOURCE_ID skip --source SOURCE`
26
+
- `sol call import resolve-facet STAGED_FILE apply --source SOURCE`
27
+
- `sol call import resolve-facet STAGED_FILE skip --source SOURCE`
28
+
- `sol call import resolve-config FIELD apply --source SOURCE`
29
+
- `sol call import resolve-config FIELD keep --source SOURCE`
30
+
- `sol call import resolve-config-all --source SOURCE --category transferable`
31
+
- `sol call import resolve-config-all --source SOURCE --category preference`
32
+
33
+
## Process
34
+
35
+
### Step 1: Confirm Source
36
+
37
+
The source name must be provided as input when you are invoked. If it is missing, ask for it before doing anything else.
38
+
39
+
### Step 2: Check All Staged Work
40
+
41
+
Run:
42
+
43
+
```bash
44
+
sol call import list-staged --source SOURCE
45
+
```
46
+
47
+
If nothing is staged, report `Nothing to review` and exit.
48
+
49
+
### Step 3: Review Entities
50
+
51
+
Process all staged entities using the same decision logic as the entity review workflow:
52
+
- merge when the match is clearly correct
53
+
- create when the entity is valid but distinct
54
+
- skip only when the staged record should be discarded
55
+
56
+
### Step 4: Review Facets
57
+
58
+
Only start facet review after entity staging is clear. Then process staged facet items using the facet review workflow.
59
+
60
+
### Step 5: Review Config
61
+
62
+
After entities and facets are resolved, review config differences:
63
+
- batch-apply transferable fields
64
+
- inspect preference fields individually
65
+
66
+
### Step 6: Final Verification
67
+
68
+
Run `sol call import list-staged --source SOURCE` again and confirm whether anything remains staged.
69
+
70
+
### Step 7: Report
71
+
72
+
Report a complete summary:
73
+
- entities merged / created / skipped
74
+
- facet items applied / skipped
75
+
- config fields applied / kept
76
+
- any remaining staged items or blockers
77
+
78
+
## Quality Guidelines
79
+
80
+
### DO:
81
+
82
+
- Follow the dependency order strictly
83
+
- Clear each area before moving to the next when feasible
84
+
- Surface blockers immediately and explicitly
85
+
- End with a final verification pass
86
+
87
+
### DON'T:
88
+
89
+
- Attempt facet resolution before entity review is complete
90
+
- Finish without checking the staged queues again
91
+
- Leave unresolved work without naming what remains
+87
apps/import/talent/review_config.md
+87
apps/import/talent/review_config.md
···
1
+
{
2
+
"type": "cogitate",
3
+
"title": "Import Config Reviewer",
4
+
"description": "Reviews and resolves staged config field differences from journal-to-journal import.",
5
+
"color": "#6a1b9a",
6
+
"group": "Import"
7
+
}
8
+
9
+
$sol_identity
10
+
11
+
$facets
12
+
13
+
## Core Mission
14
+
15
+
Review config field differences between source and target journals and decide which values to apply.
16
+
17
+
## Tooling
18
+
19
+
- `sol call import list-staged --source SOURCE --area config` - list config diff as JSON
20
+
- `sol call import resolve-config FIELD apply --source SOURCE` - apply one config field from source
21
+
- `sol call import resolve-config FIELD keep --source SOURCE` - keep the local target value for one field
22
+
- `sol call import resolve-config-all --source SOURCE --category transferable` - batch-apply all transferable fields
23
+
- `sol call import resolve-config-all --source SOURCE --category preference` - batch-apply all preference fields
24
+
25
+
## Process
26
+
27
+
### Step 1: Confirm Source
28
+
29
+
The source name must be provided as input when you are invoked. If it is missing, ask for it first.
30
+
31
+
### Step 2: List Config Diff
32
+
33
+
Run:
34
+
35
+
```bash
36
+
sol call import list-staged --source SOURCE --area config
37
+
```
38
+
39
+
If nothing is returned, report `No config differences to review` and exit.
40
+
41
+
### Step 3: Apply Transferable Fields
42
+
43
+
Transferable fields represent identity values such as name, bio, pronouns, aliases, email addresses, and timezone. These should usually follow the owner across journals.
44
+
45
+
Run:
46
+
47
+
```bash
48
+
sol call import resolve-config-all --source SOURCE --category transferable
49
+
```
50
+
51
+
### Step 4: Review Preference Fields
52
+
53
+
For each remaining preference field:
54
+
55
+
- If target is empty and source has a value, usually apply
56
+
- If both have values, compare source and target and make a judgment call
57
+
- When a local preference should remain, keep it explicitly
58
+
59
+
Use:
60
+
61
+
```bash
62
+
sol call import resolve-config FIELD apply --source SOURCE
63
+
sol call import resolve-config FIELD keep --source SOURCE
64
+
```
65
+
66
+
### Step 5: Report
67
+
68
+
Report:
69
+
- How many transferable fields were applied in batch
70
+
- How many preference fields were applied
71
+
- How many preference fields were kept
72
+
- Whether any config diff remains
73
+
74
+
## Quality Guidelines
75
+
76
+
### DO:
77
+
78
+
- Apply transferable identity fields by default
79
+
- Review remaining preference fields explicitly
80
+
- Mention both source and target values when a preference choice is non-obvious
81
+
- Confirm when the config staging queue is empty
82
+
83
+
### DON'T:
84
+
85
+
- Skip config review when a diff exists
86
+
- Keep a local value without considering whether the source should follow the owner
87
+
- Apply preference changes blindly when they conflict with an intentional local setup
+92
apps/import/talent/review_entities.md
+92
apps/import/talent/review_entities.md
···
1
+
{
2
+
"type": "cogitate",
3
+
"title": "Import Entity Reviewer",
4
+
"description": "Reviews and resolves staged entities from journal-to-journal import, handling low-confidence matches, ID collisions, and principal conflicts.",
5
+
"color": "#1565c0",
6
+
"group": "Import"
7
+
}
8
+
9
+
$sol_identity
10
+
11
+
$facets
12
+
13
+
## Core Mission
14
+
15
+
Review staged entities left by journal import and resolve each one: merge into an existing entity, create a new entity, or skip it.
16
+
17
+
## Tooling
18
+
19
+
- `sol call import list-staged --source SOURCE --area entities` - list staged entities as JSONL
20
+
- `sol call import resolve-entity SOURCE_ID merge --source SOURCE --target TARGET_ID` - merge into an existing entity
21
+
- `sol call import resolve-entity SOURCE_ID create --source SOURCE` - create a new entity
22
+
- `sol call import resolve-entity SOURCE_ID skip --source SOURCE` - discard the staged entity
23
+
24
+
## Process
25
+
26
+
### Step 1: Confirm Source
27
+
28
+
The source name must be provided as input when you are invoked. If it is missing, ask for it before doing anything else.
29
+
30
+
### Step 2: List Staged Entities
31
+
32
+
Run:
33
+
34
+
```bash
35
+
sol call import list-staged --source SOURCE --area entities
36
+
```
37
+
38
+
Parse the JSONL output and process every staged entity in the batch.
39
+
40
+
### Step 3: Decide Per Staged Entity
41
+
42
+
- **low_confidence_match**: Compare `source_entity` with `match_candidates`. Look at name similarity, type, aka values, and email overlap. If the candidate is clearly the same logical entity, merge. If not, create.
43
+
- **id_collision**: The source slug collides with a different entity in the target journal. Compare names and types. Merge if they are the same entity. Otherwise create; the CLI will allocate a new slug when needed.
44
+
- **principal_conflict**: The source entity is marked principal but the target journal already has one. Usually create; the CLI will strip `is_principal` if necessary. Only merge if the match clearly points to the same logical person.
45
+
46
+
### Step 4: Execute Resolutions
47
+
48
+
Use exactly one resolution command per staged entity:
49
+
50
+
- Merge:
51
+
52
+
```bash
53
+
sol call import resolve-entity SOURCE_ID merge --source SOURCE --target TARGET_ID
54
+
```
55
+
56
+
- Create:
57
+
58
+
```bash
59
+
sol call import resolve-entity SOURCE_ID create --source SOURCE
60
+
```
61
+
62
+
- Skip:
63
+
64
+
```bash
65
+
sol call import resolve-entity SOURCE_ID skip --source SOURCE
66
+
```
67
+
68
+
### Step 5: Verify and Report
69
+
70
+
Run `sol call import list-staged --source SOURCE --area entities` again to confirm the queue is clear or identify any remaining items.
71
+
72
+
Report:
73
+
- How many entities were merged
74
+
- How many were created
75
+
- How many were skipped
76
+
- Any staged entities still remaining and why
77
+
78
+
## Quality Guidelines
79
+
80
+
### DO:
81
+
82
+
- Process all staged entities in one pass when feasible
83
+
- Prefer merge when the staged entity is clearly the same logical entity
84
+
- Explain why each merge/create/skip choice was reasonable
85
+
- Use create instead of skip when the entity appears valid but unmatched
86
+
87
+
### DON'T:
88
+
89
+
- Leave staged entities unresolved without saying why
90
+
- Merge clearly different people, companies, projects, or tools
91
+
- Skip a valid new entity just because it needs a new slug
92
+
- Assume a principal conflict means the entity should be discarded
+93
apps/import/talent/review_facets.md
+93
apps/import/talent/review_facets.md
···
1
+
{
2
+
"type": "cogitate",
3
+
"title": "Import Facet Reviewer",
4
+
"description": "Reviews and resolves staged facet items from journal-to-journal import, handling unmapped entities and facet.json conflicts.",
5
+
"color": "#2e7d32",
6
+
"group": "Import"
7
+
}
8
+
9
+
$sol_identity
10
+
11
+
$facets
12
+
13
+
## Core Mission
14
+
15
+
Review staged facet items and resolve them. Entity review must complete first because unmapped facet items depend on the entity `id_map`.
16
+
17
+
## Tooling
18
+
19
+
- `sol call import list-staged --source SOURCE --area facets` - list staged facet items as JSONL
20
+
- `sol call import resolve-facet STAGED_FILE apply --source SOURCE` - apply a staged facet item
21
+
- `sol call import resolve-facet STAGED_FILE skip --source SOURCE` - discard a staged facet item
22
+
- `sol call import list-staged --source SOURCE --area entities` - check whether entity review is complete
23
+
24
+
## Process
25
+
26
+
### Step 1: Confirm Source
27
+
28
+
The source name must be provided as input when you are invoked. If it is missing, ask for it first.
29
+
30
+
### Step 2: Check Entity Review Status
31
+
32
+
Run:
33
+
34
+
```bash
35
+
sol call import list-staged --source SOURCE --area entities
36
+
```
37
+
38
+
If any entity items remain staged, stop immediately and report:
39
+
40
+
`Entity review must complete first. X entities still staged.`
41
+
42
+
### Step 3: List Staged Facet Items
43
+
44
+
Run:
45
+
46
+
```bash
47
+
sol call import list-staged --source SOURCE --area facets
48
+
```
49
+
50
+
Parse the JSONL output and review each staged item.
51
+
52
+
### Step 4: Decide Per Staged Item
53
+
54
+
- **unmapped_entity**: If entity review is complete, apply the staged file. If the CLI still reports a missing mapping, surface that dependency clearly.
55
+
- **facet_json_conflict**: Compare `source_content` and `target_content`. Apply source when it looks newer, richer, or more complete. Keep target when the local version is clearly preferred. If the difference is ambiguous, prefer applying source.
56
+
57
+
### Step 5: Execute Resolutions
58
+
59
+
- Apply:
60
+
61
+
```bash
62
+
sol call import resolve-facet STAGED_FILE apply --source SOURCE
63
+
```
64
+
65
+
- Skip:
66
+
67
+
```bash
68
+
sol call import resolve-facet STAGED_FILE skip --source SOURCE
69
+
```
70
+
71
+
### Step 6: Verify and Report
72
+
73
+
Run `sol call import list-staged --source SOURCE --area facets` again to confirm what remains.
74
+
75
+
Report:
76
+
- How many facet items were applied
77
+
- How many were skipped
78
+
- Any blocked items and the missing dependency or reason
79
+
80
+
## Quality Guidelines
81
+
82
+
### DO:
83
+
84
+
- Enforce the entity-review dependency before resolving facet items
85
+
- Apply unmapped entity files once the mapping exists
86
+
- Compare both sides of a `facet_json_conflict` before deciding
87
+
- Surface exactly which staged file failed if resolution is blocked
88
+
89
+
### DON'T:
90
+
91
+
- Start facet resolution while entity staging is still unresolved
92
+
- Skip an unmapped entity file just because it depends on entity review
93
+
- Ignore the content difference in `facet_json_conflict`
+635
tests/test_import_call.py
+635
tests/test_import_call.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 importlib import import_module
8
+
from pathlib import Path
9
+
10
+
import pytest
11
+
from typer.testing import CliRunner
12
+
13
+
import convey.state
14
+
import think.utils
15
+
from think.call import call_app
16
+
from think.entities.journal import (
17
+
clear_journal_entity_cache,
18
+
load_journal_entity,
19
+
save_journal_entity,
20
+
)
21
+
from think.entities.relationships import load_facet_relationship
22
+
23
+
journal_sources = import_module("apps.import.journal_sources")
24
+
25
+
create_state_directory = journal_sources.create_state_directory
26
+
generate_key = journal_sources.generate_key
27
+
get_state_directory = journal_sources.get_state_directory
28
+
save_journal_source = journal_sources.save_journal_source
29
+
30
+
runner = CliRunner()
31
+
32
+
33
+
def _source(name="test-source", key=None, **overrides):
34
+
if key is None:
35
+
key = generate_key()
36
+
source = {
37
+
"name": name,
38
+
"key": key,
39
+
"created_at": 1000,
40
+
"enabled": True,
41
+
"revoked": False,
42
+
"revoked_at": None,
43
+
"stats": {
44
+
"segments_received": 0,
45
+
"entities_received": 0,
46
+
"facets_received": 0,
47
+
"imports_received": 0,
48
+
"config_received": 0,
49
+
},
50
+
}
51
+
source.update(overrides)
52
+
return source
53
+
54
+
55
+
@pytest.fixture
56
+
def import_env(tmp_path, monkeypatch):
57
+
"""Set up a temp journal with an import source and state directory."""
58
+
59
+
monkeypatch.setattr(convey.state, "journal_root", str(tmp_path), raising=False)
60
+
monkeypatch.setenv("_SOLSTONE_JOURNAL_OVERRIDE", str(tmp_path))
61
+
think.utils._journal_path_cache = None
62
+
clear_journal_entity_cache()
63
+
(tmp_path / "apps" / "import" / "journal_sources").mkdir(parents=True, exist_ok=True)
64
+
(tmp_path / "config").mkdir(parents=True, exist_ok=True)
65
+
66
+
key = generate_key()
67
+
source = _source(key=key)
68
+
save_journal_source(source)
69
+
key_prefix = key[:8]
70
+
create_state_directory(tmp_path, key_prefix)
71
+
72
+
return {
73
+
"root": tmp_path,
74
+
"key": key,
75
+
"key_prefix": key_prefix,
76
+
"source": source,
77
+
}
78
+
79
+
80
+
def _write_json(path: Path, data: dict) -> None:
81
+
path.parent.mkdir(parents=True, exist_ok=True)
82
+
path.write_text(json.dumps(data, indent=2, ensure_ascii=False) + "\n", encoding="utf-8")
83
+
84
+
85
+
def _read_json(path: Path) -> dict:
86
+
return json.loads(path.read_text(encoding="utf-8"))
87
+
88
+
89
+
def _read_log(path: Path) -> list[dict]:
90
+
if not path.exists():
91
+
return []
92
+
return [
93
+
json.loads(line)
94
+
for line in path.read_text(encoding="utf-8").splitlines()
95
+
if line.strip()
96
+
]
97
+
98
+
99
+
def _write_entity_state(key_prefix: str, state: dict) -> None:
100
+
state_path = get_state_directory(key_prefix) / "entities" / "state.json"
101
+
_write_json(state_path, state)
102
+
103
+
104
+
def test_list_staged_empty_state(import_env):
105
+
result = runner.invoke(call_app, ["import", "list-staged", "--source", "test-source"])
106
+
107
+
assert result.exit_code == 0
108
+
assert result.stdout.strip() == ""
109
+
110
+
111
+
def test_list_staged_with_staged_entities(import_env):
112
+
staged_path = (
113
+
get_state_directory(import_env["key_prefix"])
114
+
/ "entities"
115
+
/ "staged"
116
+
/ "test-entity.json"
117
+
)
118
+
_write_json(
119
+
staged_path,
120
+
{
121
+
"source_entity": {"id": "test-entity", "name": "Test Entity", "type": "Tool"},
122
+
"match_candidates": [{"id": "target-id", "name": "Target Entity", "tier": 8}],
123
+
"reason": "low_confidence_match",
124
+
"staged_at": "2026-04-14T00:00:00+00:00",
125
+
},
126
+
)
127
+
128
+
result = runner.invoke(
129
+
call_app,
130
+
["import", "list-staged", "--source", "test-source", "--area", "entities"],
131
+
)
132
+
133
+
assert result.exit_code == 0
134
+
lines = [json.loads(line) for line in result.stdout.splitlines() if line.strip()]
135
+
assert lines == [
136
+
{
137
+
"area": "entities",
138
+
"source_id": "test-entity",
139
+
"reason": "low_confidence_match",
140
+
"source_entity": {"id": "test-entity", "name": "Test Entity", "type": "Tool"},
141
+
"match_candidates": [{"id": "target-id", "name": "Target Entity", "tier": 8}],
142
+
"staged_at": "2026-04-14T00:00:00+00:00",
143
+
}
144
+
]
145
+
146
+
147
+
def test_list_staged_with_config_diff(import_env):
148
+
diff_path = get_state_directory(import_env["key_prefix"]) / "config" / "diff.json"
149
+
_write_json(
150
+
diff_path,
151
+
{
152
+
"identity.name": {
153
+
"source": "Remote User",
154
+
"target": "Local User",
155
+
"category": "transferable",
156
+
}
157
+
},
158
+
)
159
+
160
+
result = runner.invoke(
161
+
call_app,
162
+
["import", "list-staged", "--source", "test-source", "--area", "config"],
163
+
)
164
+
165
+
assert result.exit_code == 0
166
+
lines = [json.loads(line) for line in result.stdout.splitlines() if line.strip()]
167
+
assert lines == [
168
+
{
169
+
"area": "config",
170
+
"diff": {
171
+
"identity.name": {
172
+
"source": "Remote User",
173
+
"target": "Local User",
174
+
"category": "transferable",
175
+
}
176
+
},
177
+
}
178
+
]
179
+
180
+
181
+
def test_list_staged_with_staged_facets(import_env):
182
+
staged_file = "personal/entity_relationship/entities__source_entity__entity.json.staged.json"
183
+
staged_path = get_state_directory(import_env["key_prefix"]) / "facets" / "staged" / staged_file
184
+
_write_json(
185
+
staged_path,
186
+
{
187
+
"reason": "unmapped_entity",
188
+
"source_entity_id": "source_entity",
189
+
"explanation": "Entity 'source_entity' has no mapping in entities/state.json id_map",
190
+
"source_path": "entities/source_entity/entity.json",
191
+
"source_data": json.dumps({"entity_id": "source_entity"}, ensure_ascii=False, indent=2)
192
+
+ "\n",
193
+
"staged_at": "2026-04-14T00:00:00+00:00",
194
+
},
195
+
)
196
+
197
+
result = runner.invoke(
198
+
call_app,
199
+
["import", "list-staged", "--source", "test-source", "--area", "facets"],
200
+
)
201
+
202
+
assert result.exit_code == 0
203
+
lines = [json.loads(line) for line in result.stdout.splitlines() if line.strip()]
204
+
assert lines == [
205
+
{
206
+
"area": "facets",
207
+
"staged_file": staged_file,
208
+
"facet": "personal",
209
+
"file_type": "entity_relationship",
210
+
"reason": "unmapped_entity",
211
+
"source_entity_id": "source_entity",
212
+
"explanation": "Entity 'source_entity' has no mapping in entities/state.json id_map",
213
+
"source_path": "entities/source_entity/entity.json",
214
+
"source_data": json.dumps({"entity_id": "source_entity"}, ensure_ascii=False, indent=2)
215
+
+ "\n",
216
+
"staged_at": "2026-04-14T00:00:00+00:00",
217
+
}
218
+
]
219
+
220
+
221
+
def test_resolve_entity_merge(import_env):
222
+
save_journal_entity(
223
+
{
224
+
"id": "target-id",
225
+
"name": "Alice Johnson",
226
+
"type": "Person",
227
+
"aka": ["Ali"],
228
+
"emails": ["alice@old.com"],
229
+
"created_at": 2000,
230
+
}
231
+
)
232
+
staged_path = (
233
+
get_state_directory(import_env["key_prefix"])
234
+
/ "entities"
235
+
/ "staged"
236
+
/ "test-entity.json"
237
+
)
238
+
_write_json(
239
+
staged_path,
240
+
{
241
+
"source_entity": {
242
+
"id": "test-entity",
243
+
"name": "Alice Johnson",
244
+
"type": "Person",
245
+
"aka": ["AJ"],
246
+
"emails": ["alice@new.com"],
247
+
"created_at": 1000,
248
+
},
249
+
"match_candidates": [{"id": "target-id", "name": "Alice Johnson", "tier": 8}],
250
+
"reason": "low_confidence_match",
251
+
"staged_at": "2026-04-14T00:00:00+00:00",
252
+
},
253
+
)
254
+
255
+
result = runner.invoke(
256
+
call_app,
257
+
[
258
+
"import",
259
+
"resolve-entity",
260
+
"test-entity",
261
+
"merge",
262
+
"--source",
263
+
"test-source",
264
+
"--target",
265
+
"target-id",
266
+
],
267
+
)
268
+
269
+
assert result.exit_code == 0
270
+
assert not staged_path.exists()
271
+
merged = load_journal_entity("target-id")
272
+
assert merged is not None
273
+
assert merged["aka"] == ["AJ", "Ali"]
274
+
assert merged["emails"] == ["alice@old.com", "alice@new.com"]
275
+
assert merged["created_at"] == 1000
276
+
277
+
state = _read_json(get_state_directory(import_env["key_prefix"]) / "entities" / "state.json")
278
+
assert state["id_map"]["test-entity"] == "target-id"
279
+
280
+
log_entries = _read_log(get_state_directory(import_env["key_prefix"]) / "entities" / "log.jsonl")
281
+
assert log_entries[-1]["action"] == "resolved_merge"
282
+
assert log_entries[-1]["resolved_by"] == "talent"
283
+
284
+
285
+
def test_resolve_entity_create(import_env):
286
+
save_journal_entity({"id": "test-entity", "name": "Occupied Entity", "type": "Tool"})
287
+
staged_path = (
288
+
get_state_directory(import_env["key_prefix"])
289
+
/ "entities"
290
+
/ "staged"
291
+
/ "test-entity.json"
292
+
)
293
+
_write_json(
294
+
staged_path,
295
+
{
296
+
"source_entity": {"id": "test-entity", "name": "Fresh Entity", "type": "Tool"},
297
+
"match_candidates": [{"id": "test-entity", "name": "Occupied Entity", "tier": None}],
298
+
"reason": "id_collision",
299
+
"staged_at": "2026-04-14T00:00:00+00:00",
300
+
},
301
+
)
302
+
303
+
result = runner.invoke(
304
+
call_app,
305
+
["import", "resolve-entity", "test-entity", "create", "--source", "test-source"],
306
+
)
307
+
308
+
assert result.exit_code == 0
309
+
assert not staged_path.exists()
310
+
created = load_journal_entity("fresh_entity")
311
+
assert created is not None
312
+
assert created["name"] == "Fresh Entity"
313
+
314
+
state = _read_json(get_state_directory(import_env["key_prefix"]) / "entities" / "state.json")
315
+
assert state["id_map"]["test-entity"] == "fresh_entity"
316
+
317
+
318
+
def test_resolve_entity_create_principal_conflict(import_env):
319
+
save_journal_entity(
320
+
{
321
+
"id": "existing-principal",
322
+
"name": "Existing Principal",
323
+
"type": "Person",
324
+
"is_principal": True,
325
+
}
326
+
)
327
+
staged_path = (
328
+
get_state_directory(import_env["key_prefix"])
329
+
/ "entities"
330
+
/ "staged"
331
+
/ "new-principal.json"
332
+
)
333
+
_write_json(
334
+
staged_path,
335
+
{
336
+
"source_entity": {
337
+
"id": "new-principal",
338
+
"name": "New Principal",
339
+
"type": "Person",
340
+
"is_principal": True,
341
+
},
342
+
"match_candidates": [],
343
+
"reason": "principal_conflict",
344
+
"staged_at": "2026-04-14T00:00:00+00:00",
345
+
},
346
+
)
347
+
348
+
result = runner.invoke(
349
+
call_app,
350
+
["import", "resolve-entity", "new-principal", "create", "--source", "test-source"],
351
+
)
352
+
353
+
assert result.exit_code == 0
354
+
created = load_journal_entity("new-principal")
355
+
assert created is not None
356
+
assert created["is_principal"] is False
357
+
358
+
359
+
def test_resolve_entity_skip(import_env):
360
+
staged_path = (
361
+
get_state_directory(import_env["key_prefix"])
362
+
/ "entities"
363
+
/ "staged"
364
+
/ "test-entity.json"
365
+
)
366
+
_write_json(
367
+
staged_path,
368
+
{
369
+
"source_entity": {"id": "test-entity", "name": "Skip Entity", "type": "Tool"},
370
+
"match_candidates": [],
371
+
"reason": "principal_conflict",
372
+
"staged_at": "2026-04-14T00:00:00+00:00",
373
+
},
374
+
)
375
+
376
+
result = runner.invoke(
377
+
call_app,
378
+
["import", "resolve-entity", "test-entity", "skip", "--source", "test-source"],
379
+
)
380
+
381
+
assert result.exit_code == 0
382
+
assert not staged_path.exists()
383
+
assert load_journal_entity("test-entity") is None
384
+
385
+
log_entries = _read_log(get_state_directory(import_env["key_prefix"]) / "entities" / "log.jsonl")
386
+
assert log_entries[-1]["action"] == "resolved_skip"
387
+
assert log_entries[-1]["resolved_by"] == "talent"
388
+
389
+
390
+
def test_resolve_config_apply(import_env):
391
+
diff_path = get_state_directory(import_env["key_prefix"]) / "config" / "diff.json"
392
+
_write_json(
393
+
diff_path,
394
+
{
395
+
"identity.name": {
396
+
"source": "Remote User",
397
+
"target": "Local User",
398
+
"category": "transferable",
399
+
}
400
+
},
401
+
)
402
+
_write_json(
403
+
get_state_directory(import_env["key_prefix"]) / "config" / "source_config.json",
404
+
{"identity": {"name": "Remote User"}},
405
+
)
406
+
_write_json(import_env["root"] / "config" / "journal.json", {"identity": {"name": "Local User"}})
407
+
408
+
result = runner.invoke(
409
+
call_app,
410
+
["import", "resolve-config", "identity.name", "apply", "--source", "test-source"],
411
+
)
412
+
413
+
assert result.exit_code == 0
414
+
journal_config = _read_json(import_env["root"] / "config" / "journal.json")
415
+
assert journal_config["identity"]["name"] == "Remote User"
416
+
assert not diff_path.exists()
417
+
418
+
log_entries = _read_log(get_state_directory(import_env["key_prefix"]) / "config" / "log.jsonl")
419
+
assert log_entries[-1]["action"] == "config_field_applied"
420
+
assert log_entries[-1]["resolved_by"] == "talent"
421
+
422
+
423
+
def test_resolve_config_keep(import_env):
424
+
diff_path = get_state_directory(import_env["key_prefix"]) / "config" / "diff.json"
425
+
_write_json(
426
+
diff_path,
427
+
{
428
+
"retention.days": {
429
+
"source": 30,
430
+
"target": 90,
431
+
"category": "preference",
432
+
}
433
+
},
434
+
)
435
+
_write_json(
436
+
get_state_directory(import_env["key_prefix"]) / "config" / "source_config.json",
437
+
{"retention": {"days": 30}},
438
+
)
439
+
_write_json(import_env["root"] / "config" / "journal.json", {"retention": {"days": 90}})
440
+
441
+
result = runner.invoke(
442
+
call_app,
443
+
["import", "resolve-config", "retention.days", "keep", "--source", "test-source"],
444
+
)
445
+
446
+
assert result.exit_code == 0
447
+
journal_config = _read_json(import_env["root"] / "config" / "journal.json")
448
+
assert journal_config["retention"]["days"] == 90
449
+
assert not diff_path.exists()
450
+
451
+
452
+
def test_resolve_config_all_transferable(import_env):
453
+
diff_path = get_state_directory(import_env["key_prefix"]) / "config" / "diff.json"
454
+
_write_json(
455
+
diff_path,
456
+
{
457
+
"identity.name": {
458
+
"source": "Remote User",
459
+
"target": "Local User",
460
+
"category": "transferable",
461
+
},
462
+
"retention.days": {
463
+
"source": 30,
464
+
"target": 90,
465
+
"category": "preference",
466
+
},
467
+
},
468
+
)
469
+
_write_json(
470
+
get_state_directory(import_env["key_prefix"]) / "config" / "source_config.json",
471
+
{
472
+
"identity": {"name": "Remote User"},
473
+
"retention": {"days": 30},
474
+
},
475
+
)
476
+
_write_json(
477
+
import_env["root"] / "config" / "journal.json",
478
+
{"identity": {"name": "Local User"}, "retention": {"days": 90}},
479
+
)
480
+
481
+
result = runner.invoke(
482
+
call_app,
483
+
[
484
+
"import",
485
+
"resolve-config-all",
486
+
"--source",
487
+
"test-source",
488
+
"--category",
489
+
"transferable",
490
+
],
491
+
)
492
+
493
+
assert result.exit_code == 0
494
+
journal_config = _read_json(import_env["root"] / "config" / "journal.json")
495
+
assert journal_config["identity"]["name"] == "Remote User"
496
+
assert journal_config["retention"]["days"] == 90
497
+
498
+
remaining_diff = _read_json(diff_path)
499
+
assert list(remaining_diff) == ["retention.days"]
500
+
501
+
502
+
def test_resolve_facet_apply_unmapped_entity(import_env):
503
+
_write_entity_state(
504
+
import_env["key_prefix"],
505
+
{"id_map": {"source_entity": "target_entity"}, "received": {}},
506
+
)
507
+
staged_file = "personal/entity_relationship/entities__source_entity__entity.json.staged.json"
508
+
staged_path = get_state_directory(import_env["key_prefix"]) / "facets" / "staged" / staged_file
509
+
_write_json(
510
+
staged_path,
511
+
{
512
+
"reason": "unmapped_entity",
513
+
"source_entity_id": "source_entity",
514
+
"explanation": "Entity 'source_entity' has no mapping in entities/state.json id_map",
515
+
"source_path": "entities/source_entity/entity.json",
516
+
"source_data": json.dumps(
517
+
{"entity_id": "source_entity", "description": "imported relationship"},
518
+
ensure_ascii=False,
519
+
indent=2,
520
+
)
521
+
+ "\n",
522
+
"staged_at": "2026-04-14T00:00:00+00:00",
523
+
},
524
+
)
525
+
526
+
result = runner.invoke(
527
+
call_app,
528
+
["import", "resolve-facet", staged_file, "apply", "--source", "test-source"],
529
+
)
530
+
531
+
assert result.exit_code == 0
532
+
assert not staged_path.exists()
533
+
relationship = load_facet_relationship("personal", "target_entity")
534
+
assert relationship is not None
535
+
assert relationship["entity_id"] == "target_entity"
536
+
assert relationship["description"] == "imported relationship"
537
+
538
+
log_entries = _read_log(get_state_directory(import_env["key_prefix"]) / "facets" / "log.jsonl")
539
+
assert log_entries[-1]["action"] == "resolved_apply"
540
+
assert log_entries[-1]["resolved_by"] == "talent"
541
+
542
+
543
+
def test_resolve_facet_apply_facet_json_conflict(import_env):
544
+
target_path = import_env["root"] / "facets" / "personal" / "facet.json"
545
+
_write_json(target_path, {"title": "Local"})
546
+
staged_file = "personal/facet_json/facet.json.staged.json"
547
+
staged_path = get_state_directory(import_env["key_prefix"]) / "facets" / "staged" / staged_file
548
+
_write_json(
549
+
staged_path,
550
+
{
551
+
"reason": "facet_json_conflict",
552
+
"source_content": {"title": "Remote"},
553
+
"target_content": {"title": "Local"},
554
+
"staged_at": "2026-04-14T00:00:00+00:00",
555
+
},
556
+
)
557
+
558
+
result = runner.invoke(
559
+
call_app,
560
+
["import", "resolve-facet", staged_file, "apply", "--source", "test-source"],
561
+
)
562
+
563
+
assert result.exit_code == 0
564
+
assert not staged_path.exists()
565
+
assert _read_json(target_path) == {"title": "Remote"}
566
+
567
+
log_entries = _read_log(get_state_directory(import_env["key_prefix"]) / "facets" / "log.jsonl")
568
+
assert log_entries[-1]["action"] == "resolved_apply"
569
+
assert log_entries[-1]["item_id"] == "personal/facet.json"
570
+
assert log_entries[-1]["resolved_by"] == "talent"
571
+
572
+
573
+
def test_resolve_facet_unmapped_entity_fails_without_mapping(import_env):
574
+
staged_file = "personal/entity_relationship/entities__source_entity__entity.json.staged.json"
575
+
staged_path = get_state_directory(import_env["key_prefix"]) / "facets" / "staged" / staged_file
576
+
_write_json(
577
+
staged_path,
578
+
{
579
+
"reason": "unmapped_entity",
580
+
"source_entity_id": "source_entity",
581
+
"explanation": "Entity 'source_entity' has no mapping in entities/state.json id_map",
582
+
"source_path": "entities/source_entity/entity.json",
583
+
"source_data": json.dumps({"entity_id": "source_entity"}, ensure_ascii=False, indent=2)
584
+
+ "\n",
585
+
"staged_at": "2026-04-14T00:00:00+00:00",
586
+
},
587
+
)
588
+
589
+
result = runner.invoke(
590
+
call_app,
591
+
["import", "resolve-facet", staged_file, "apply", "--source", "test-source"],
592
+
)
593
+
594
+
assert result.exit_code == 1
595
+
assert "Entity source_entity has no mapping yet. Run entity review first." in result.stderr
596
+
assert staged_path.exists()
597
+
598
+
599
+
def test_resolve_facet_skip(import_env):
600
+
staged_file = "personal/entity_relationship/entities__source_entity__entity.json.staged.json"
601
+
staged_path = get_state_directory(import_env["key_prefix"]) / "facets" / "staged" / staged_file
602
+
_write_json(
603
+
staged_path,
604
+
{
605
+
"reason": "unmapped_entity",
606
+
"source_entity_id": "source_entity",
607
+
"explanation": "Entity 'source_entity' has no mapping in entities/state.json id_map",
608
+
"source_path": "entities/source_entity/entity.json",
609
+
"source_data": json.dumps({"entity_id": "source_entity"}, ensure_ascii=False, indent=2)
610
+
+ "\n",
611
+
"staged_at": "2026-04-14T00:00:00+00:00",
612
+
},
613
+
)
614
+
615
+
result = runner.invoke(
616
+
call_app,
617
+
["import", "resolve-facet", staged_file, "skip", "--source", "test-source"],
618
+
)
619
+
620
+
assert result.exit_code == 0
621
+
assert not staged_path.exists()
622
+
assert load_facet_relationship("personal", "source_entity") is None
623
+
624
+
625
+
def test_resolve_source_not_found(import_env):
626
+
result = runner.invoke(
627
+
call_app,
628
+
["import", "list-staged", "--source", "nonexistent"],
629
+
)
630
+
631
+
assert result.exit_code == 1
632
+
assert (
633
+
"Import source 'nonexistent' not found. Check available sources in "
634
+
"~/.local/share/solstone/app-storage/import/journal_sources/."
635
+
) in result.stderr