asobi 0.4.1

A persistent, project-local knowledge graph CLI for AI agents.
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177

#!/usr/bin/env python3
"""Use-case scenarios showing Asobi's performance and correctness for task boarding."""

import json
import os
import subprocess
import tempfile
import time
from pathlib import Path

ROOT = Path(__file__).resolve().parents[1]
BIN = ROOT / "target" / "debug" / "asobi"


def run_cmd(args: list[str], db_path: str) -> str:
    env = os.environ.copy()
    env["ASOBI_DATABASE_URL"] = db_path
    env["ASOBI_HOME"] = str(Path(db_path).parent)

    start = time.perf_counter()
    result = subprocess.run(
        [str(BIN), *args], cwd=ROOT, env=env, text=True, capture_output=True, check=True
    )
    elapsed = time.perf_counter() - start
    # Print the command and elapsed time for performance tracing
    cmd_str = f"asobi {' '.join(args)}"
    print(f"[PERF] {cmd_str:<60} | {elapsed * 1000:6.2f} ms")
    return result.stdout


def main() -> None:
    # Build the binary first to make sure it's up to date
    print("Building target/debug/asobi...")
    subprocess.run(["cargo", "build"], cwd=ROOT, check=True)

    with tempfile.TemporaryDirectory(prefix="asobi-usecase-") as tmp_dir:
        db_path = str(Path(tmp_dir) / "usecase.db")
        print(f"\n--- Initializing Usecase DB at {db_path} ---")

        # -------------------------------------------------------------
        # Use-Case 1: Seed Epic and Tasks (Using new --obs and truths)
        # -------------------------------------------------------------
        print("\n--- Use-Case 1: Seeding Epic and Tasks atomically ---")

        # Seed the epic
        run_cmd(
            ["new", "asobi:v0.3", "task", "--obs", "Epic for v0.3 release"], db_path
        )
        run_cmd(["truth", "asobi:v0.3", "status", "IN_PROGRESS"], db_path)

        # Seed 3 tasks in the epic with initial statuses and observations
        run_cmd(
            [
                "new",
                "asobi:v0.3:task-1",
                "task",
                "--obs",
                "Implement atomic bulk backup",
                "asobi:v0.3:task-2",
                "task",
                "--obs",
                "Optimize index page caching",
                "asobi:v0.3:task-3",
                "task",
                "--obs",
                "Write end-to-end integration tests",
            ],
            db_path,
        )

        # Set task statuses and blockages
        run_cmd(["truth", "asobi:v0.3:task-1", "status", "READY_TO_DISPATCH"], db_path)
        run_cmd(["truth", "asobi:v0.3:task-2", "status", "BLOCKED"], db_path)
        run_cmd(
            ["truth", "asobi:v0.3:task-2", "blocked_on", "asobi:v0.3:task-1"], db_path
        )
        run_cmd(["truth", "asobi:v0.3:task-3", "status", "READY_TO_DISPATCH"], db_path)

        # Link tasks to the epic
        run_cmd(
            [
                "link",
                "asobi:v0.3:task-1",
                "asobi:v0.3",
                "part_of",
                "asobi:v0.3:task-2",
                "asobi:v0.3",
                "part_of",
                "asobi:v0.3:task-3",
                "asobi:v0.3",
                "part_of",
            ],
            db_path,
        )

        # -------------------------------------------------------------
        # Use-Case 2: Board Search (Search by truth status only)
        # -------------------------------------------------------------
        print("\n--- Use-Case 2: Board Search (status=READY_TO_DISPATCH) ---")

        stdout = run_cmd(["search", "--where", "status=READY_TO_DISPATCH"], db_path)
        graph = json.loads(stdout)

        matched_names = {
            e["name"]
            for e in graph["entities"]
            if e.get("truths", {}).get("status") == "READY_TO_DISPATCH"
        }
        print(f"Matched READY_TO_DISPATCH entities: {matched_names}")
        assert matched_names == {"asobi:v0.3:task-1", "asobi:v0.3:task-3"}, (
            f"Mismatch: {matched_names}"
        )

        # -------------------------------------------------------------
        # Use-Case 3: Dispatch and Update Status
        # -------------------------------------------------------------
        print("\n--- Use-Case 3: Claim and complete task-1 ---")

        # Claim task-1
        run_cmd(["truth", "asobi:v0.3:task-1", "status", "IN_PROGRESS"], db_path)
        run_cmd(["obs", "asobi:v0.3:task-1", "Assigned to agent-antigravity"], db_path)

        # Complete task-1
        run_cmd(["truth", "asobi:v0.3:task-1", "status", "DONE"], db_path)
        run_cmd(
            [
                "obs",
                "asobi:v0.3:task-1",
                "Completed migration check, code is fully functional.",
            ],
            db_path,
        )

        # Verify status is updated
        stdout_show = run_cmd(["show", "asobi:v0.3:task-1"], db_path)
        show_graph = json.loads(stdout_show)
        entity = show_graph["entities"][0]
        assert entity["truths"]["status"] == "DONE"
        assert (
            "Completed migration check, code is fully functional."
            in entity["observations"]
        )

        # -------------------------------------------------------------
        # Use-Case 4: Skills Registry (Install, List, and Show)
        # -------------------------------------------------------------
        print("\n--- Use-Case 4: Skills Registry (Install, List, and Show) ---")

        # Create a mock local skill
        skills_src_dir = Path(tmp_dir) / "mock-skills"
        skills_src_dir.mkdir()
        skill_file = skills_src_dir / "test-skill.md"
        skill_file.write_text(
            "---\n"
            "name: test-skill\n"
            "description: A mock skill for use-case validation\n"
            "---\n"
            "This is the body of the test skill."
        )

        # Install the mock skill
        run_cmd(["skills", "install", str(skills_src_dir), "--all"], db_path)

        # List all skills
        list_stdout = run_cmd(["skills"], db_path)
        assert "test-skill" in list_stdout
        assert "A mock skill for use-case validation" in list_stdout

        # Show specific skill body
        show_stdout = run_cmd(["skills", "show", "test-skill"], db_path)
        assert "This is the body of the test skill." in show_stdout

        print("\nAll integration use-cases successfully validated!")


if __name__ == "__main__":
    main()