diff --git a/tests/test_git_delivery.py b/tests/test_git_delivery.py index 9ca4462..0519b20 100644 --- a/tests/test_git_delivery.py +++ b/tests/test_git_delivery.py @@ -134,6 +134,33 @@ def test_clear_removes_evidence_and_records_a_history_baseline(self): self.assertEqual(ledger.rows(), []) self.assertEqual(ledger.baseline_at(), 200) + def test_coalesce_preserves_canonical_conflicts_and_moves_legacy_evidence(self): + with tempfile.TemporaryDirectory() as tmp: + ledger = meter.GitDeliveryLedger( + str(Path(tmp) / "delivery.sqlite3"), "test-salt", + ) + ledger.record("legacy", "shared", 100, 8, 2) + ledger.record("legacy", "legacy-only", 100, 6, 1) + ledger.record("canonical", "shared", 100, 3, 4) + ledger.map_project("legacy-project", "legacy") + ledger.set_repository_coverage("legacy", True, 100, partial=True) + ledger.set_repository_coverage("canonical", False, 200, partial=False) + + ledger.coalesce_repository("legacy", "canonical") + + self.assertEqual(ledger.rows(), [ + {"repo_key": "canonical", "object_key": "legacy-only", "observed_at": 100, + "day": "1970-01-01", "added": 6, "deleted": 1}, + {"repo_key": "canonical", "object_key": "shared", "observed_at": 100, + "day": "1970-01-01", "added": 3, "deleted": 4}, + ]) + self.assertEqual(ledger.repo_key_for_project("legacy-project"), "canonical") + self.assertTrue(ledger.has_seen("canonical", "legacy-only")) + self.assertTrue(ledger.has_seen("canonical", "shared")) + self.assertEqual(ledger.repository_coverage("canonical"), { + "measured": True, "partial": True, "checked_at": 200, + }) + class GitDeliveryScannerTests(unittest.TestCase): def test_subprocess_runner_supplies_a_system_path_for_launch_agents(self): @@ -211,6 +238,119 @@ def test_actual_git_push_is_observed_locally_without_network_or_gh(self): self.assertNotIn("gh", json.dumps(result)) self.assertNotIn(str(repo), json.dumps(result)) + def test_linked_worktree_preserves_the_main_worktree_ledger_key(self): + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + repo = root / "repo" + linked = root / "linked-worktree" + subprocess.run(["git", "init", "-q", str(repo)], check=True) + subprocess.run(["git", "-C", str(repo), "config", "user.name", "Alice"], check=True) + subprocess.run([ + "git", "-C", str(repo), "config", "user.email", "alice@example.com", + ], check=True) + (repo / "README").write_text("seed\n", encoding="utf-8") + subprocess.run(["git", "-C", str(repo), "add", "README"], check=True) + subprocess.run(["git", "-C", str(repo), "commit", "-qm", "seed"], check=True) + subprocess.run([ + "git", "-C", str(repo), "worktree", "add", "-q", "-b", "linked", str(linked), + ], check=True) + service = meter.GitDeliveryService( + str(root / "delivery.sqlite3"), now=lambda: local_timestamp("2026-09-04"), + salt="test-salt", + ) + + service.scan([ + {"root": str(repo), "project": "repo · 111111"}, + {"root": str(linked), "project": "linked-worktree · 222222"}, + ]) + + main_top = subprocess.check_output( + ["git", "-C", str(repo), "rev-parse", "--show-toplevel"], text=True, + ).strip() + main_key = service.ledger.repo_key_for_project(service._hash(str(repo))) + linked_key = service.ledger.repo_key_for_project(service._hash(str(linked))) + self.assertEqual(main_key, service._hash(main_top)) + self.assertEqual(linked_key, main_key) + + def test_separate_git_dir_preserves_the_worktree_ledger_key(self): + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + repo = root / "repo" + metadata = root / "metadata" / ".git" + metadata.parent.mkdir() + subprocess.run([ + "git", "init", "-q", "--separate-git-dir", str(metadata), str(repo), + ], check=True) + subprocess.run([ + "git", "-C", str(repo), "config", "user.email", "alice@example.com", + ], check=True) + service = meter.GitDeliveryService( + str(root / "delivery.sqlite3"), now=lambda: local_timestamp("2026-09-04"), + salt="test-salt", + ) + + service.scan([{"root": str(repo), "project": "repo · 111111"}]) + + top_level = subprocess.check_output( + ["git", "-C", str(repo), "rev-parse", "--show-toplevel"], text=True, + ).strip() + repo_key = service.ledger.repo_key_for_project(service._hash(str(repo))) + self.assertEqual(repo_key, service._hash(top_level)) + + def test_linked_worktree_coalesces_legacy_evidence_without_double_counting(self): + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + repo = root / "repo" + linked = root / "linked-worktree" + subprocess.run(["git", "init", "-q", str(repo)], check=True) + subprocess.run(["git", "-C", str(repo), "config", "user.name", "Alice"], check=True) + subprocess.run([ + "git", "-C", str(repo), "config", "user.email", "alice@example.com", + ], check=True) + (repo / "README").write_text("seed\n", encoding="utf-8") + subprocess.run(["git", "-C", str(repo), "add", "README"], check=True) + subprocess.run(["git", "-C", str(repo), "commit", "-qm", "seed"], check=True) + subprocess.run([ + "git", "-C", str(repo), "worktree", "add", "-q", "-b", "linked", str(linked), + ], check=True) + service = meter.GitDeliveryService( + str(root / "delivery.sqlite3"), now=lambda: local_timestamp("2026-09-04"), + salt="test-salt", + ) + linked_top = subprocess.check_output( + ["git", "-C", str(linked), "rev-parse", "--show-toplevel"], text=True, + ).strip() + main_top = subprocess.check_output( + ["git", "-C", str(repo), "rev-parse", "--show-toplevel"], text=True, + ).strip() + legacy_key = service._hash(linked_top) + canonical_key = service._hash(main_top) + object_key = service._hash("seed-object") + label = "linked-worktree · 111111" + service.ledger.record(legacy_key, object_key, local_timestamp("2026-09-03"), 6, 2) + service.ledger.map_project(service._hash(str(linked)), legacy_key) + service.ledger.set_repository_coverage( + legacy_key, True, local_timestamp("2026-09-03"), + ) + + service.scan([{"root": str(linked), "project": label}]) + service.ledger.set_repository_coverage( + canonical_key, True, local_timestamp("2026-09-04"), + ) + payload = service.query( + label, "7", [], [label], [{"root": str(linked), "project": label}], + ) + + self.assertEqual(service.ledger.rows(), [{ + "repo_key": canonical_key, "object_key": object_key, + "observed_at": local_timestamp("2026-09-03"), "day": "2026-09-03", + "added": 6, "deleted": 2, + }]) + self.assertFalse(service.ledger.record( + canonical_key, object_key, local_timestamp("2026-09-03"), 60, 20, + )) + self.assertEqual(payload["overall"]["changed_lines"], 8) + def test_scan_uses_only_local_read_only_git_commands_and_matching_identity(self): first_oid = "a" * 40 second_oid = "b" * 40 @@ -221,6 +361,8 @@ def runner(argv, **_kwargs): args = tuple(argv[argv.index("-C") + 2:]) if args == ("rev-parse", "--show-toplevel"): return {"returncode": 0, "stdout": "/repo\n"} + if args == ("rev-parse", "--git-common-dir"): + return {"returncode": 0, "stdout": ".git\n"} if args == ("config", "--get", "user.email"): return {"returncode": 0, "stdout": "Alice@Example.com\n"} if args[0] == "for-each-ref": @@ -259,6 +401,8 @@ def runner(argv, **_kwargs): args = tuple(argv[argv.index("-C") + 2:]) if args == ("rev-parse", "--show-toplevel"): return {"returncode": 0, "stdout": "/repo\n"} + if args == ("rev-parse", "--git-common-dir"): + return {"returncode": 0, "stdout": ".git\n"} if args == ("config", "--get", "user.email"): return {"returncode": 1, "stdout": ""} if args == ("config", "--global", "--get", "user.email"): @@ -283,6 +427,8 @@ def runner(argv, **_kwargs): args = tuple(argv[argv.index("-C") + 2:]) if args == ("rev-parse", "--show-toplevel"): return {"returncode": 0, "stdout": "/repo\n"} + if args == ("rev-parse", "--git-common-dir"): + return {"returncode": 0, "stdout": ".git\n"} if args == ("config", "--get", "user.email"): return {"returncode": 0, "stdout": "alice@example.com\n"} if args[0] == "for-each-ref": @@ -322,6 +468,8 @@ def runner(argv, **_kwargs): args = tuple(argv[argv.index("-C") + 2:]) if args == ("rev-parse", "--show-toplevel"): return {"returncode": 0, "stdout": "/repo\n"} + if args == ("rev-parse", "--git-common-dir"): + return {"returncode": 0, "stdout": ".git\n"} if args == ("config", "--get", "user.email"): return {"returncode": 0, "stdout": "alice@example.com\n"} if args[0] == "for-each-ref": @@ -604,9 +752,12 @@ def test_install_bootstrap_scans_while_the_invoking_app_has_repository_access(se "ok": True, "new_changed_lines": 12, "coverage": {"measured": 1}, } service.project_suffix.return_value = "a1b2c3" + service.repository_key.return_value = "opaque-repository-key" with mock.patch.object(meter, "all_session_sources", return_value=sources), mock.patch.object( meter, "git_delivery_service", return_value=service, - ): + ), mock.patch.object(meter.os.path, "isdir", return_value=True), mock.patch.object( + meter, "publish_git_delivery_candidates", + ) as publish_candidates: result = meter.bootstrap_git_delivery() self.assertEqual(result["new_changed_lines"], 12) @@ -615,6 +766,7 @@ def test_install_bootstrap_scans_while_the_invoking_app_has_repository_access(se service.scan.call_args.args[0][0]["root"], "/Users/alice/Code/private-project", ) + publish_candidates.assert_not_called() def test_each_installer_bootstraps_delivery_before_server_start(self): contracts = ( @@ -629,27 +781,309 @@ def test_each_installer_bootstraps_delivery_before_server_start(self): ) self.assertLess(installer.index("bootstrap_git_delivery"), installer.index(start_marker)) - def test_candidates_are_bounded_and_never_project_absolute_paths(self): + def test_candidates_skip_missing_non_git_and_linked_worktrees_before_the_budget(self): with tempfile.TemporaryDirectory() as tmp: service = meter.GitDeliveryService( str(Path(tmp) / "delivery.sqlite3"), salt="test-salt", ) - root = "/Users/alice/Code/private-project" + root = Path(tmp) / "private-project" + linked = Path(tmp) / "linked-worktree" + missing = Path(tmp) / "removed-project" + non_repository = Path(tmp) / "notes" + subprocess.run(["git", "init", "-q", str(root)], check=True) + subprocess.run(["git", "-C", str(root), "config", "user.name", "Alice"], check=True) + subprocess.run([ + "git", "-C", str(root), "config", "user.email", "alice@example.com", + ], check=True) + (root / "README").write_text("seed\n", encoding="utf-8") + subprocess.run(["git", "-C", str(root), "add", "README"], check=True) + subprocess.run(["git", "-C", str(root), "commit", "-qm", "seed"], check=True) + subprocess.run([ + "git", "-C", str(root), "worktree", "add", "-q", "-b", "linked", str(linked), + ], check=True) + non_repository.mkdir() with mock.patch.object(meter, "git_delivery_service", return_value=service): candidates = meter.git_delivery_candidates([ - {"project": root}, - {"project": root}, - {"project": "not-a-root"}, + {"project": str(linked)}, + {"project": str(missing)}, + {"project": str(non_repository)}, + {"project": str(root)}, ]) self.assertEqual(len(candidates), 1) - self.assertEqual(candidates[0]["root"], root) - self.assertRegex(candidates[0]["project"], r"^private-project · [0-9a-f]{6}$") - self.assertNotIn("/Users/alice", candidates[0]["project"]) - self.assertTrue(candidates[0]["project"].endswith(service.project_suffix(root))) + self.assertEqual(candidates[0]["root"], str(linked)) + self.assertRegex(candidates[0]["project"], r"^linked-worktree · [0-9a-f]{6}$") + self.assertNotIn(str(Path(tmp)), candidates[0]["project"]) + self.assertTrue(candidates[0]["project"].endswith(service.project_suffix(linked))) self.assertFalse(candidates[0]["project"].endswith( - hashlib.sha256(root.encode("utf-8")).hexdigest()[:6] + hashlib.sha256(str(linked).encode("utf-8")).hexdigest()[:6] )) + self.assertEqual(candidates[0]["aliases"][0]["root"], str(root)) + labels, canonical_by_source, _repo_by_label = service._project_repo_mapping( + [candidates[0]["project"]], candidates, + ) + self.assertEqual(labels, [candidates[0]["project"]]) + self.assertEqual( + canonical_by_source[candidates[0]["aliases"][0]["project"]], + candidates[0]["project"], + ) + + def test_git_delivery_state_hides_the_unscanned_overflow_project(self): + candidates = [ + {"root": "/repo-{}".format(index), "project": "repo-{}".format(index)} + for index in range(git_delivery.MAX_REPOSITORIES + 1) + ] + service = mock.Mock() + service.query.return_value = {"ok": True} + inventory = { + "ready": True, "sources": (), "count": 0, "clients": {}, "updated_at": 1, + "revision": 1, "git_delivery_candidates": tuple(candidates), + "git_delivery_candidates_revision": 1, + } + with mock.patch.dict(meter._xsess, {"data": {}, "internal_rows": []}), mock.patch.object( + meter, "_SOURCE_INVENTORY", inventory, + ), mock.patch.object(meter, "git_delivery_service", return_value=service): + meter.git_delivery_state() + + self.assertEqual( + service.query.call_args.args[3], + sorted("repo-{}".format(index) for index in range(git_delivery.MAX_REPOSITORIES)), + ) + self.assertEqual( + service.query.call_args.args[4], tuple(candidates[:git_delivery.MAX_REPOSITORIES]), + ) + + def test_git_delivery_state_reuses_scan_owned_candidates_without_discovery(self): + candidates = [{"root": "/repo", "project": "repo · 111111", "repo_key": "opaque"}] + service = mock.Mock() + service.query.return_value = {"ok": True} + inventory = { + "ready": True, "sources": (), "count": 0, "clients": {}, "updated_at": 1, + "revision": 1, "git_delivery_candidates": tuple(candidates), + "git_delivery_candidates_revision": 1, + } + with mock.patch.dict(meter._xsess, {"data": {}, "internal_rows": []}), mock.patch.object( + meter, "_SOURCE_INVENTORY", inventory, + ), mock.patch.object(meter, "git_delivery_candidates", return_value=[]) as discover, mock.patch.object( + meter, "git_delivery_service", return_value=service, + ): + meter.git_delivery_state() + meter.git_delivery_state() + + discover.assert_not_called() + self.assertEqual(service.query.call_args.args[3], ["repo · 111111"]) + + def test_git_delivery_state_uses_an_empty_snapshot_before_the_first_scan(self): + service = mock.Mock() + service.query.return_value = {"ok": True} + inventory = { + "ready": True, "sources": ({"project": "/repo"},), "count": 1, + "clients": {}, "updated_at": 1, "revision": 1, + "git_delivery_candidates": (), "git_delivery_candidates_revision": None, + } + with mock.patch.dict(meter._xsess, {"data": {}, "internal_rows": []}), mock.patch.object( + meter, "_SOURCE_INVENTORY", inventory, + ), mock.patch.object(meter, "git_delivery_candidates", return_value=[]) as discover, mock.patch.object( + meter, "git_delivery_service", return_value=service, + ): + meter.git_delivery_state() + + discover.assert_not_called() + self.assertEqual(service.query.call_args.args[3:], ([], ())) + + def test_git_delivery_watcher_publishes_the_scanned_candidate_snapshot(self): + candidates = [{"root": "/repo", "project": "repo · 111111", "repo_key": "opaque"}] + inventory = { + "ready": True, "sources": ({"project": "/repo"},), "count": 1, + "clients": {}, "updated_at": 1, "revision": 4, + "git_delivery_source_signature": "snapshot", + "git_delivery_candidates": (), "git_delivery_candidates_revision": None, + } + service = mock.Mock() + with mock.patch.object(meter, "_SOURCE_INVENTORY", inventory), mock.patch.object( + meter, "git_delivery_candidates", return_value=candidates, + ) as discover, mock.patch.object(meter, "git_delivery_service", return_value=service), mock.patch.object( + meter, "publish_git_delivery_candidates", + ) as publish_candidates, mock.patch.object( + meter.time, "monotonic", side_effect=[0.0, 0.0, 0.0], + ), mock.patch.object(meter.time, "sleep", side_effect=StopIteration): + with self.assertRaises(StopIteration): + meter.git_delivery_watcher() + + discover.assert_called_once_with(inventory["sources"]) + service.scan.assert_called_once_with(candidates) + publish_candidates.assert_called_once_with(candidates, 4, "snapshot") + + def test_source_inventory_mtime_refresh_keeps_git_candidates_until_project_roots_change(self): + candidates = ({"root": "/repo", "project": "repo · 111111", "repo_key": "opaque"},) + initial = { + "ready": True, "sources": (), "count": 0, "clients": {}, "updated_at": 0, + "revision": 0, "git_delivery_candidates": (), "git_delivery_candidates_revision": None, + } + wake = mock.Mock() + with mock.patch.object(meter, "_SOURCE_INVENTORY", initial), mock.patch.object( + meter, "_git_delivery_wake", wake, + ): + meter.publish_source_inventory([{"project": "/repo", "mtime": 1}]) + meter.publish_git_delivery_candidates(candidates) + wake.reset_mock() + preserved = meter.publish_source_inventory([{"project": "/repo", "mtime": 2}]) + wake.assert_not_called() + service = mock.Mock() + service.query.return_value = {"ok": True} + with mock.patch.dict(meter._xsess, {"data": {}, "internal_rows": []}), mock.patch.object( + meter, "git_delivery_candidates", return_value=(), + ) as discover, mock.patch.object(meter, "git_delivery_service", return_value=service): + meter.git_delivery_state() + discover.assert_not_called() + self.assertEqual(service.query.call_args.args[3], ["repo · 111111"]) + changed = meter.publish_source_inventory([{"project": "/other", "mtime": 2}]) + + self.assertEqual(preserved["git_delivery_candidates"], candidates) + self.assertEqual( + preserved["git_delivery_candidates_revision"], preserved["revision"], + ) + self.assertEqual(changed["git_delivery_candidates"], ()) + self.assertIsNone(changed["git_delivery_candidates_revision"]) + self.assertGreater(changed["revision"], preserved["revision"]) + wake.set.assert_called_once_with() + + def test_candidates_keep_later_aliases_for_an_eligible_repository_after_overflow(self): + with tempfile.TemporaryDirectory() as tmp: + root = Path(tmp) + primary = root / "primary" + alias = root / "primary-alias" + others = [root / "repo-{}".format(index) for index in range( + git_delivery.MAX_REPOSITORIES, + )] + for path in [primary, alias, *others]: + path.mkdir() + keys = {str(primary): "primary-key", str(alias): "primary-key"} + keys.update({str(path): "repo-key-{}".format(index) for index, path in enumerate(others)}) + service = mock.Mock() + service.repository_key.side_effect = lambda path: keys[path] + service.project_suffix.side_effect = lambda path: "{:06x}".format( + abs(hash(str(path))) % 0x1000000, + ) + alias_label = "primary-alias · {:06x}".format( + abs(hash(str(alias))) % 0x1000000, + ) + sources = ([{"project": str(primary)}] + + [{"project": str(path)} for path in others] + + [{"project": str(alias)}]) + with mock.patch.object(meter, "git_delivery_service", return_value=service): + candidates = meter.git_delivery_candidates(sources) + + self.assertEqual(len(candidates), git_delivery.MAX_REPOSITORIES + 1) + self.assertEqual(candidates[0]["root"], str(primary)) + self.assertEqual(candidates[0].get("aliases"), [{ + "root": str(alias), + "project": alias_label, + }]) + + def test_candidates_deduplicate_exact_roots_before_repository_resolution(self): + service = mock.Mock() + service.repository_key.side_effect = ["shared", "shared"] + service.project_suffix.return_value = "111111" + sources = [ + {"project": "/repo/main"}, + {"project": "/repo/main"}, + {"project": "/repo/linked"}, + ] + with mock.patch.object(meter, "git_delivery_service", return_value=service), mock.patch.object( + meter.os.path, "isdir", return_value=True, + ): + candidates = meter.git_delivery_candidates(sources) + + self.assertEqual(service.repository_key.call_args_list, [ + mock.call("/repo/main"), mock.call("/repo/linked"), + ]) + self.assertEqual(candidates[0]["root"], "/repo/main") + self.assertEqual(candidates[0]["aliases"][0]["root"], "/repo/linked") + + def test_watcher_floors_repeated_wakes_after_a_prompt_ready_scan(self): + service = mock.Mock() + service.scan.side_effect = [None, RuntimeError("stop watcher")] + wake = mock.Mock() + first_sources = ({"project": "/repo/first"},) + latest_sources = ({"project": "/repo/latest"},) + inventory = { + "ready": True, "sources": first_sources, "revision": 1, + "git_delivery_source_signature": "first", + } + + def release_floor(seconds): + self.assertEqual(seconds, meter.GIT_DELIVERY_INTERVAL_S) + inventory.update({"sources": latest_sources, "revision": 2, + "git_delivery_source_signature": "latest"}) + + with mock.patch.object(meter, "_SOURCE_INVENTORY", inventory), mock.patch.object( + meter, "_git_delivery_wake", wake, + ), mock.patch.object( + meter, "git_delivery_candidates", side_effect=[ + [{"root": "/repo/first", "project": "first", "repo_key": "first"}], + [{"root": "/repo/latest", "project": "latest", "repo_key": "latest"}], + ], + ) as candidates, mock.patch.object(meter, "git_delivery_service", return_value=service), mock.patch.object( + meter, "publish_git_delivery_candidates", + ), mock.patch.object( + meter.time, "monotonic", side_effect=[0.0, 0.0, 0.0, 300.0], + ), mock.patch.object(meter.time, "sleep", side_effect=release_floor) as sleep: + with self.assertRaisesRegex(RuntimeError, "stop watcher"): + meter.git_delivery_watcher() + + self.assertEqual(candidates.call_args_list, [ + mock.call(first_sources), mock.call(latest_sources), + ]) + sleep.assert_called_once_with(meter.GIT_DELIVERY_INTERVAL_S) + wake.wait.assert_not_called() + + def test_ordered_project_roots_invalidate_candidates_when_the_cap_can_change(self): + roots = ["/repo-{}".format(index) for index in range(git_delivery.MAX_REPOSITORIES + 1)] + sources = [{"project": root} for root in roots] + service = mock.Mock() + service.repository_key.side_effect = lambda root: "key-{}".format(root) + service.project_suffix.return_value = "111111" + initial = { + "ready": True, "sources": (), "count": 0, "clients": {}, "updated_at": 0, + "revision": 0, "git_delivery_source_signature": "", "git_delivery_candidates": (), + "git_delivery_candidates_revision": None, + } + wake = mock.Mock() + with mock.patch.object(meter, "_SOURCE_INVENTORY", initial), mock.patch.object( + meter, "_git_delivery_wake", wake, + ), mock.patch.object(meter, "git_delivery_service", return_value=service), mock.patch.object( + meter.os.path, "isdir", return_value=True, + ): + first_candidates = meter.git_delivery_candidates(sources) + meter.publish_source_inventory(sources) + meter.publish_git_delivery_candidates(first_candidates) + wake.reset_mock() + reversed_sources = list(reversed(sources)) + reversed_candidates = meter.git_delivery_candidates(reversed_sources) + changed = meter.publish_source_inventory(reversed_sources) + + self.assertEqual(first_candidates[0]["root"], roots[0]) + self.assertEqual(reversed_candidates[0]["root"], roots[-1]) + self.assertEqual(changed["git_delivery_candidates"], ()) + wake.set.assert_called_once_with() + + def test_stale_watcher_publish_cannot_clobber_newer_inventory(self): + initial = { + "ready": True, "sources": (), "count": 0, "clients": {}, "updated_at": 0, + "revision": 1, "git_delivery_source_signature": "first", "git_delivery_candidates": (), + "git_delivery_candidates_revision": None, + } + wake = mock.Mock() + candidates = [{"root": "/repo", "project": "repo", "repo_key": "opaque"}] + with mock.patch.object(meter, "_SOURCE_INVENTORY", initial), mock.patch.object( + meter, "_git_delivery_wake", wake, + ): + meter.publish_source_inventory([{"project": "/other"}]) + published = meter.publish_git_delivery_candidates(candidates, 1, "first") + + self.assertFalse(published) + self.assertEqual(meter._SOURCE_INVENTORY["git_delivery_candidates"], ()) def test_clear_uses_service_baseline_before_waking_the_watcher(self): service = mock.Mock() diff --git a/token_meter/app.py b/token_meter/app.py index 1e159f4..7c24ad4 100644 --- a/token_meter/app.py +++ b/token_meter/app.py @@ -103,7 +103,9 @@ tool_identity as _domain_tool_identity, tool_summary as _domain_tool_summary, ) -from token_meter.services.git_delivery import GitDeliveryLedger, GitDeliveryService +from token_meter.services.git_delivery import ( + GitDeliveryLedger, GitDeliveryService, MAX_QUERY_PROJECTS, MAX_REPOSITORIES, +) from token_meter.models.catalog import ( ANTHROPIC_PRICE as CLAUDE_PRICE, BUILTIN_MODEL_PRICE_HISTORY as _CANONICAL_BUILTIN_MODEL_PRICE_HISTORY, @@ -344,7 +346,12 @@ def hermes_state_db_path(environ=None): "count": None, "clients": {}, "updated_at": None, + "revision": 0, + "git_delivery_source_signature": "", + "git_delivery_candidates": (), + "git_delivery_candidates_revision": None, } +_source_inventory_lock = threading.Lock() _git_delivery_service_instance = None _git_delivery_service_lock = threading.Lock() _git_delivery_wake = threading.Event() @@ -3048,6 +3055,36 @@ def supported_runtime_phrase(): return "{}, or {}".format(", ".join(labels[:-1]), labels[-1]) +def git_delivery_project_roots(sources): + """Return ordered unique normalized roots relevant to Git candidate selection.""" + roots = [] + seen = set() + for source in sources or (): + raw_project = source.get("project") if isinstance(source, dict) else "" + if not isinstance(raw_project, str): + continue + if project_filter_key(raw_project) == OTHER_LOCAL_SESSIONS_PROJECT: + continue + root = os.path.abspath(os.path.expanduser(raw_project)) + if root not in seen: + roots.append(root) + seen.add(root) + return tuple(roots) + + +def git_delivery_source_signature(roots): + """Return a private ordered-root membership signature for Git candidates.""" + return hashlib.sha256( + "\0".join(roots).encode("utf-8", "replace"), + ).hexdigest() + + +def source_inventory_snapshot(): + """Read one atomically published source inventory reference.""" + with _source_inventory_lock: + return _SOURCE_INVENTORY + + def publish_source_inventory(sources): """Atomically publish a reusable discovery snapshot for lightweight endpoints.""" global _SOURCE_INVENTORY @@ -3065,20 +3102,35 @@ def publish_source_inventory(sources): clients = defaultdict(int) for source in source_rows: clients[source.get("client") or source.get("provider") or "unknown"] += 1 - _SOURCE_INVENTORY = { - "ready": True, - "sources": source_rows, - "count": len(source_rows), - "clients": dict(clients), - "updated_at": time.time(), - } - _git_delivery_wake.set() + roots = git_delivery_project_roots(source_rows) + signature = git_delivery_source_signature(roots) + with _source_inventory_lock: + previous = _SOURCE_INVENTORY + candidates_changed = signature != previous.get("git_delivery_source_signature") + revision = int(previous.get("revision") or 0) + int(candidates_changed) + _SOURCE_INVENTORY = { + "ready": True, + "sources": source_rows, + "count": len(source_rows), + "clients": dict(clients), + "updated_at": time.time(), + "revision": revision, + "git_delivery_source_signature": signature, + "git_delivery_candidates": ( + () if candidates_changed else previous.get("git_delivery_candidates") or () + ), + "git_delivery_candidates_revision": ( + None if candidates_changed else previous.get("git_delivery_candidates_revision") + ), + } + if candidates_changed: + _git_delivery_wake.set() return _SOURCE_INVENTORY def cached_session_sources(): """Return the watcher-owned source snapshot without touching the filesystem.""" - inventory = _SOURCE_INVENTORY + inventory = source_inventory_snapshot() return list(inventory.get("sources") or ()), bool(inventory.get("ready")) @@ -6926,22 +6978,93 @@ def delivery_project_label(value): def git_delivery_candidates(sources=None): """Derive bounded repository candidates from already-discovered projects.""" candidates = [] + candidates_by_repository = {} + seen_repositories = set() seen_roots = set() + alias_count = 0 + service = git_delivery_service() for source in list(sources or ()): raw_project = source.get("project") if isinstance(source, dict) else "" if project_filter_key(raw_project) == OTHER_LOCAL_SESSIONS_PROJECT: continue root = os.path.abspath(os.path.expanduser(raw_project)) - project = delivery_project_label(raw_project) - if len(root) > 4096 or not project or root in seen_roots: + if len(root) > 4096 or root in seen_roots or not os.path.isdir(root): continue - candidates.append({"root": root, "project": project}) seen_roots.add(root) - if len(candidates) >= 50: - break + repository_key = service.repository_key(root) + if not repository_key: + continue + project = delivery_project_label(raw_project) + if not project: + continue + existing = candidates_by_repository.get(repository_key) + if existing is not None: + aliases = existing.setdefault("aliases", []) + if ( + alias_count < MAX_QUERY_PROJECTS + and not any(alias.get("root") == root for alias in aliases) + ): + aliases.append({"root": root, "project": project}) + alias_count += 1 + continue + if repository_key in seen_repositories or len(candidates) >= MAX_REPOSITORIES + 1: + continue + candidate = {"root": root, "project": project, "repo_key": repository_key} + candidates.append(candidate) + candidates_by_repository[repository_key] = candidate + seen_repositories.add(repository_key) return candidates +def publish_git_delivery_candidates(candidates, revision=None, signature=None): + """Atomically publish a bounded private candidate snapshot after a Git scan.""" + global _SOURCE_INVENTORY + rows = [] + for candidate in tuple(candidates or ())[:MAX_REPOSITORIES + 1]: + if not isinstance(candidate, dict): + continue + root = candidate.get("root") + project = candidate.get("project") + repo_key = candidate.get("repo_key") + if not all(isinstance(value, str) and value for value in (root, project, repo_key)): + continue + aliases = [] + for alias in tuple(candidate.get("aliases") or ())[:MAX_QUERY_PROJECTS]: + if not isinstance(alias, dict): + continue + alias_root = alias.get("root") + alias_project = alias.get("project") + if isinstance(alias_root, str) and alias_root and isinstance(alias_project, str) and alias_project: + aliases.append({"root": alias_root, "project": alias_project}) + row = {"root": root, "project": project, "repo_key": repo_key} + if aliases: + row["aliases"] = tuple(aliases) + rows.append(row) + with _source_inventory_lock: + inventory = _SOURCE_INVENTORY + revision = inventory.get("revision") if revision is None else revision + signature = inventory.get("git_delivery_source_signature") if signature is None else signature + if ( + revision != inventory.get("revision") + or signature != inventory.get("git_delivery_source_signature") + ): + return False + _SOURCE_INVENTORY = dict( + inventory, + git_delivery_candidates=tuple(rows), + git_delivery_candidates_revision=revision, + ) + return True + + +def cached_git_delivery_candidates(): + """Return the watcher-owned Git candidate snapshot without invoking Git.""" + inventory = source_inventory_snapshot() + if inventory.get("git_delivery_candidates_revision") != inventory.get("revision"): + return () + return inventory.get("git_delivery_candidates") or () + + def delivery_spend_rows(internal_rows): """Project daily cost and token totals into the content-free Git boundary.""" rows = [] @@ -7050,7 +7173,8 @@ def git_delivery_service(): def bootstrap_git_delivery(): """Seed readable local push history from the interactive installer context.""" sources = all_session_sources() - return git_delivery_service().scan(git_delivery_candidates(sources)) + candidates = git_delivery_candidates(sources) + return git_delivery_service().scan(candidates) def git_delivery_state(project="", range_key="7"): @@ -7058,14 +7182,15 @@ def git_delivery_state(project="", range_key="7"): if _xsess.get("data") is None: cross_session() internal_rows = _xsess.get("internal_rows") or () - candidates = git_delivery_candidates(_SOURCE_INVENTORY.get("sources") or ()) - projects = sorted({candidate["project"] for candidate in candidates}) + candidates = cached_git_delivery_candidates() + eligible = candidates[:MAX_REPOSITORIES] + projects = sorted({candidate["project"] for candidate in eligible}) return git_delivery_service().query( project, range_key, delivery_spend_rows(internal_rows), projects, - candidates, + eligible, ) @@ -7080,15 +7205,26 @@ def clear_git_delivery_activity(confirm=False): def git_delivery_watcher(): """Inspect local successful-push reflogs every five minutes.""" + next_scan_at = 0.0 while True: - if not _SOURCE_INVENTORY.get("ready"): + inventory = source_inventory_snapshot() + if not inventory.get("ready"): _git_delivery_wake.wait(1.0) _git_delivery_wake.clear() continue - candidates = git_delivery_candidates(_SOURCE_INVENTORY.get("sources") or ()) + remaining = next_scan_at - time.monotonic() + if remaining > 0: + _git_delivery_wake.clear() + time.sleep(remaining) + continue + candidates = git_delivery_candidates(inventory.get("sources") or ()) git_delivery_service().scan(candidates) - _git_delivery_wake.wait(GIT_DELIVERY_INTERVAL_S) - _git_delivery_wake.clear() + publish_git_delivery_candidates( + candidates, + inventory.get("revision"), + inventory.get("git_delivery_source_signature"), + ) + next_scan_at = time.monotonic() + GIT_DELIVERY_INTERVAL_S def aggregate_model_stats(session_rows): diff --git a/token_meter/services/git_delivery.py b/token_meter/services/git_delivery.py index 485bc9c..c8163ec 100644 --- a/token_meter/services/git_delivery.py +++ b/token_meter/services/git_delivery.py @@ -287,6 +287,63 @@ def repository_coverage(self, repo_key): "checked_at": int(row["checked_at"]), } + def coalesce_repository(self, legacy_key, canonical_key): + """Atomically move hashed legacy evidence into a canonical repository key.""" + if ( + not isinstance(legacy_key, str) or not legacy_key + or not isinstance(canonical_key, str) or not canonical_key + or legacy_key == canonical_key + ): + return + with self._connect() as connection: + coverage = connection.execute( + """ + SELECT repo_key, measured, partial, checked_at + FROM delivery_repository_coverage WHERE repo_key IN (?, ?) + """, + (legacy_key, canonical_key), + ).fetchall() + connection.execute( + """ + INSERT OR IGNORE INTO delivery_observations + (repo_key, object_key, observed_at, day, added, deleted) + SELECT ?, object_key, observed_at, day, added, deleted + FROM delivery_observations WHERE repo_key = ? + """, + (canonical_key, legacy_key), + ) + connection.execute( + "DELETE FROM delivery_observations WHERE repo_key = ?", (legacy_key,)) + connection.execute( + """ + INSERT OR IGNORE INTO delivery_seen (repo_key, object_key) + SELECT ?, object_key FROM delivery_seen WHERE repo_key = ? + """, + (canonical_key, legacy_key), + ) + connection.execute( + "DELETE FROM delivery_seen WHERE repo_key = ?", (legacy_key,)) + connection.execute( + "UPDATE delivery_project_mappings SET repo_key = ? WHERE repo_key = ?", + (canonical_key, legacy_key), + ) + if coverage: + measured = any(bool(row["measured"]) for row in coverage) + partial = any(bool(row["partial"]) for row in coverage) + checked_at = max(int(row["checked_at"]) for row in coverage) + connection.execute( + """ + INSERT OR REPLACE INTO delivery_repository_coverage + (repo_key, measured, partial, checked_at) + VALUES (?, ?, ?, ?) + """, + (canonical_key, int(measured), int(partial), checked_at), + ) + connection.execute( + "DELETE FROM delivery_repository_coverage WHERE repo_key = ?", + (legacy_key,), + ) + def set_last_checked(self, value): value = self._timestamp(value) with self._connect() as connection: @@ -404,6 +461,51 @@ def project_suffix(self, value): """Return a compact per-machine opaque project discriminator.""" return self._hash(str(value or ""))[:6] + def _repository_roots(self, root): + """Return the scanned worktree root and its canonical main-worktree root.""" + if not isinstance(root, str) or not root or len(root) > 4096: + return "", "" + root = os.path.abspath(os.path.expanduser(root)) + code, output = self._run_git(root, ("rev-parse", "--show-toplevel")) + resolved = str(output or "").strip().splitlines()[0] if output else "" + if code != 0 or not os.path.isabs(resolved) or len(resolved) > 4096: + return "", "" + resolved = os.path.normpath(resolved) + code, output = self._run_git(resolved, ("rev-parse", "--git-common-dir")) + common_dir = str(output or "").strip().splitlines()[0] if output else "" + if code != 0 or not common_dir or len(common_dir) > 4096: + return "", "" + if common_dir == ".git": + return resolved, resolved + if not os.path.isabs(common_dir): + return "", "" + common_dir = os.path.normpath(common_dir) + if os.path.basename(common_dir) != ".git": + return "", "" + code, output = self._run_git(resolved, ("rev-parse", "--git-dir")) + git_dir = str(output or "").strip().splitlines()[0] if output else "" + if code != 0 or not os.path.isabs(git_dir): + return resolved, resolved + if os.path.normpath(git_dir) == common_dir: + return resolved, resolved + canonical = os.path.dirname(common_dir) + if not os.path.isdir(canonical): + return "", "" + code, output = self._run_git(canonical, ("rev-parse", "--show-toplevel")) + reported = str(output or "").strip().splitlines()[0] if output else "" + if ( + code != 0 + or not os.path.isabs(reported) + or os.path.normpath(reported) != canonical + ): + return resolved, resolved + return resolved, canonical + + def repository_key(self, root): + """Return an opaque canonical repository identity for a live local Git root.""" + _resolved, canonical = self._repository_roots(root) + return self._hash(canonical) if canonical else "" + def clear(self): """Forget observations and baseline reflog history already present.""" with self._scan_lock: @@ -498,19 +600,26 @@ def _introduced_commits(self, root, new_oid, old_oid): def _scan_candidate(self, candidate, checked_at, remaining): root = candidate.get("root") if isinstance(candidate, dict) else "" - if not isinstance(root, str) or not root or len(root) > 4096: + resolved, canonical = self._repository_roots(root) + if not resolved: return 0, 0, "repository_unavailable", False, True, remaining - code, output = self._run_git(root, ("rev-parse", "--show-toplevel")) - resolved = str(output or "").strip().splitlines()[0] if output else "" - if code != 0 or not os.path.isabs(resolved) or len(resolved) > 4096: - return 0, 0, "repository_unavailable", False, True, remaining - resolved = os.path.normpath(resolved) - repo_key = self._hash(resolved) - source_root = str(candidate.get("root") or "") - self.ledger.map_project(self._hash(source_root), repo_key) - project = candidate.get("project") if isinstance(candidate, dict) else "" - if isinstance(project, str) and project: - self._project_repo_keys[project] = repo_key + repo_key = self._hash(canonical) + rows = [candidate] + list(candidate.get("aliases") or ()) + for row in rows: + if not isinstance(row, dict): + continue + source_root = row.get("root") + if isinstance(source_root, str) and source_root: + if source_root == root: + source_resolved, source_canonical = resolved, canonical + else: + source_resolved, source_canonical = self._repository_roots(source_root) + if source_resolved and source_canonical == canonical: + self.ledger.coalesce_repository(self._hash(source_resolved), repo_key) + self.ledger.map_project(self._hash(source_root), repo_key) + project = row.get("project") + if isinstance(project, str) and project: + self._project_repo_keys[project] = repo_key if repo_key in self._active_repo_keys: return 0, 0, "coalesced", True, False, remaining self._active_repo_keys.add(repo_key) @@ -691,20 +800,24 @@ def _project_repo_mapping(self, projects, candidates): for candidate in candidates or (): if not isinstance(candidate, dict): continue - root = candidate.get("root") - project = candidate.get("project") - if not isinstance(root, str) or project not in source_projects: - continue - source_repo_keys[project] = ( - self._project_repo_keys.get(project) - or self.ledger.repo_key_for_project(self._hash(root)) - ) + rows = [candidate] + list(candidate.get("aliases") or ()) + for row in rows: + if not isinstance(row, dict): + continue + root = row.get("root") + project = row.get("project") + if not isinstance(root, str) or not isinstance(project, str) or not project: + continue + source_repo_keys[project] = ( + self._project_repo_keys.get(project) + or candidate.get("repo_key") + or self.ledger.repo_key_for_project(self._hash(root)) + ) canonical_by_repo = {} - for label, repo_key in source_repo_keys.items(): + for label in source_projects: + repo_key = source_repo_keys.get(label) if repo_key: - canonical_by_repo[repo_key] = min( - canonical_by_repo.get(repo_key, label), label, - ) + canonical_by_repo.setdefault(repo_key, label) canonical_by_source = { label: canonical_by_repo.get(repo_key, label) for label, repo_key in source_repo_keys.items()