Webapp product walkthrough¶

TakashiSasaki/templates
        |
        | Composition tooling と contracts を提供
        v
あなたの別 task-ledger product repository

separate repository
  ↓
Composition install
  ↓
doctor
  ↓
composition.json
  ↓
inspect → plan → review → apply → validate
  ↓
valid Composition scaffold
  ↓
明確な editing boundary

mkdir /absolute/path/to/task-ledger
cd /absolute/path/to/task-ledger
git init

python --version

python -I -c '
import hashlib
import pathlib
import subprocess
import sys
import tempfile
import urllib.request

url = "https://raw.githubusercontent.com/TakashiSasaki/templates/c328fbe2bf733cf32cea54c1054570a94afa693a/scripts/install_composition_skill.py"
expected = "d5422e28b29aaf015c14ffe4d17ae4a0478e0d108a98c951f978f7016f90e607"
data = urllib.request.urlopen(url, timeout=30).read()
actual = hashlib.sha256(data).hexdigest()
if actual != expected:
    raise SystemExit(f"installer SHA-256 mismatch: expected {expected}, got {actual}")
print(f"Verified Composition installer SHA-256: {actual}")
with tempfile.NamedTemporaryFile(suffix=".py", delete=False) as handle:
    handle.write(data)
    installer = pathlib.Path(handle.name)
try:
    subprocess.run([sys.executable, "-I", str(installer), *sys.argv[1:]], check=True)
finally:
    installer.unlink(missing_ok=True)
' /absolute/path/to/agent-skills/composition

python /absolute/path/to/agent-skills/composition/scripts/run.py \
  --repository /absolute/path/to/task-ledger \
  doctor

{
  "schema_version": 1,
  "recipe": "webapp",
  "components": {
    "include": [
      "capability.cli",
      "capability.runtime",
      "capability.service"
    ],
    "exclude": []
  },
  "parameters": {}
}

python /absolute/path/to/agent-skills/composition/scripts/run.py \
  --repository /absolute/path/to/task-ledger \
  inspect

python /absolute/path/to/agent-skills/composition/scripts/run.py \
  --repository /absolute/path/to/task-ledger \
  plan --config /absolute/path/to/task-ledger/composition.json

python /absolute/path/to/agent-skills/composition/scripts/run.py \
  --repository /absolute/path/to/task-ledger \
  apply --config /absolute/path/to/task-ledger/composition.json

python /absolute/path/to/agent-skills/composition/scripts/run.py \
  --repository /absolute/path/to/task-ledger \
  validate

Implementation ecosystem: CPython 3.11+
Persistence: SQLite
Server command: python -m task_ledger.cli --database task-ledger.db serve --host 127.0.0.1 --port 8080
Distribution: source execution for this example

GET    /api/tasks?status=all|open|completed
GET    /api/tasks/{id}
POST   /api/tasks
PATCH  /api/tasks/{id}
DELETE /api/tasks/{id}
GET    /healthz

{
  "$schema": "../schemas/service-interface.schema.json",
  "schemaVersion": 2,
  "mode": "product",
  "protocol": "http-json",
  "operations": [
    {
      "id": "list-tasks",
      "invocation": "GET /api/tasks?status=all|open|completed",
      "success": "200 JSON task array for a valid status filter",
      "negative": "400 JSON error for an invalid status filter"
    },
    {
      "id": "get-task",
      "invocation": "GET /api/tasks/{id}",
      "success": "200 JSON task for an existing id",
      "negative": "404 JSON error for a missing id"
    },
    {
      "id": "create-task",
      "invocation": "POST /api/tasks",
      "success": "201 JSON task for a non-empty title",
      "negative": "400 JSON error for an empty title"
    },
    {
      "id": "update-task",
      "invocation": "PATCH /api/tasks/{id}",
      "success": "200 JSON updated task for an existing id",
      "negative": "404 JSON error for a missing id"
    },
    {
      "id": "delete-task",
      "invocation": "DELETE /api/tasks/{id}",
      "success": "204 for an existing id",
      "negative": "404 JSON error when the id no longer exists"
    },
    {
      "id": "health",
      "invocation": "GET /healthz",
      "success": "200 JSON status ok",
      "negative": "404 JSON error for an unknown service path"
    }
  ]
}

python -m task_ledger.cli --database task-ledger.db list --status all
python -m task_ledger.cli --database task-ledger.db export

{
  "$schema": "../schemas/cli-interface.schema.json",
  "schemaVersion": 2,
  "mode": "product",
  "entrypoints": [
    {
      "id": "task-ledger",
      "command": ["python", "-m", "task_ledger.cli", "--database", "task-ledger.db"],
      "workingDirectory": ".",
      "helpArguments": ["--help"],
      "versionArguments": ["--version"],
      "structuredOutput": {
        "arguments": ["export"],
        "format": "json",
        "contractVersionField": "contractVersion"
      },
      "exitCodes": {
        "success": 0,
        "negativeResult": 1,
        "invalidInput": 2,
        "unavailable": 3,
        "refused": 4,
        "internalFailure": 5,
        "additionalInputRequired": 6
      }
    }
  ]
}

mkdir -p task_ledger/static tests scripts
touch task_ledger/__init__.py

from __future__ import annotations

import argparse
import json
import sqlite3
from http import HTTPStatus
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
from pathlib import Path
from urllib.parse import parse_qs, urlparse


def connect(database: str) -> sqlite3.Connection:
    connection = sqlite3.connect(database)
    connection.row_factory = sqlite3.Row
    connection.execute(
        "CREATE TABLE IF NOT EXISTS tasks ("
        "id INTEGER PRIMARY KEY, title TEXT NOT NULL, completed INTEGER NOT NULL DEFAULT 0)"
    )
    connection.commit()
    return connection


def task_dict(row: sqlite3.Row) -> dict[str, object]:
    return {"id": row["id"], "title": row["title"], "completed": bool(row["completed"])}


def list_tasks(database: str, status: str = "all") -> list[dict[str, object]]:
    if status not in {"all", "open", "completed"}:
        raise ValueError("status must be all, open, or completed")
    query = "SELECT id, title, completed FROM tasks"
    parameters: tuple[object, ...] = ()
    if status != "all":
        query += " WHERE completed = ?"
        parameters = (1 if status == "completed" else 0,)
    query += " ORDER BY id"
    with connect(database) as connection:
        return [task_dict(row) for row in connection.execute(query, parameters)]


def create_task(database: str, title: object) -> dict[str, object]:
    if not isinstance(title, str) or not title.strip():
        raise ValueError("title must be a non-empty string")
    with connect(database) as connection:
        cursor = connection.execute("INSERT INTO tasks(title) VALUES (?)", (title.strip(),))
        row = connection.execute(
            "SELECT id, title, completed FROM tasks WHERE id = ?", (cursor.lastrowid,)
        ).fetchone()
    assert row is not None
    return task_dict(row)


def get_task(database: str, task_id: int) -> dict[str, object] | None:
    with connect(database) as connection:
        row = connection.execute(
            "SELECT id, title, completed FROM tasks WHERE id = ?", (task_id,)
        ).fetchone()
    return task_dict(row) if row is not None else None


def update_task(database: str, task_id: int, changes: dict[str, object]) -> dict[str, object] | None:
    current = get_task(database, task_id)
    if current is None:
        return None
    title = changes.get("title", current["title"])
    completed = changes.get("completed", current["completed"])
    if not isinstance(title, str) or not title.strip() or not isinstance(completed, bool):
        raise ValueError("title must be non-empty and completed must be boolean")
    with connect(database) as connection:
        connection.execute(
            "UPDATE tasks SET title = ?, completed = ? WHERE id = ?",
            (title.strip(), int(completed), task_id),
        )
    return get_task(database, task_id)


def delete_task(database: str, task_id: int) -> bool:
    with connect(database) as connection:
        cursor = connection.execute("DELETE FROM tasks WHERE id = ?", (task_id,))
    return cursor.rowcount == 1


def make_server(database: str, host: str, port: int) -> ThreadingHTTPServer:
    static_root = Path(__file__).with_name("static")

    class Handler(BaseHTTPRequestHandler):
        def send_json(self, status: int, value: object) -> None:
            body = json.dumps(value).encode()
            self.send_response(status)
            self.send_header("Content-Type", "application/json; charset=utf-8")
            self.send_header("Content-Length", str(len(body)))
            self.end_headers()
            self.wfile.write(body)

        def read_json(self) -> dict[str, object]:
            length = int(self.headers.get("Content-Length", "0"))
            value = json.loads(self.rfile.read(length) or b"{}")
            if not isinstance(value, dict):
                raise ValueError("JSON body must be an object")
            return value

        def task_id(self, path: str) -> int | None:
            parts = path.strip("/").split("/")
            if len(parts) == 3 and parts[:2] == ["api", "tasks"] and parts[2].isdigit():
                return int(parts[2])
            return None

        def do_GET(self) -> None:
            parsed = urlparse(self.path)
            if parsed.path == "/healthz":
                self.send_json(HTTPStatus.OK, {"status": "ok"})
                return
            if parsed.path == "/api/tasks":
                status = parse_qs(parsed.query).get("status", ["all"])[0]
                try:
                    self.send_json(HTTPStatus.OK, list_tasks(database, status))
                except ValueError as exc:
                    self.send_json(HTTPStatus.BAD_REQUEST, {"error": str(exc)})
                return
            task_id = self.task_id(parsed.path)
            if task_id is not None:
                task = get_task(database, task_id)
                self.send_json(HTTPStatus.OK, task) if task else self.send_json(
                    HTTPStatus.NOT_FOUND, {"error": "task not found"}
                )
                return
            if parsed.path == "/":
                body = (static_root / "index.html").read_bytes()
                self.send_response(HTTPStatus.OK)
                self.send_header("Content-Type", "text/html; charset=utf-8")
                self.send_header("Content-Length", str(len(body)))
                self.end_headers()
                self.wfile.write(body)
                return
            self.send_json(HTTPStatus.NOT_FOUND, {"error": "not found"})

        def do_POST(self) -> None:
            if urlparse(self.path).path != "/api/tasks":
                self.send_json(HTTPStatus.NOT_FOUND, {"error": "not found"})
                return
            try:
                self.send_json(HTTPStatus.CREATED, create_task(database, self.read_json().get("title")))
            except (ValueError, json.JSONDecodeError) as exc:
                self.send_json(HTTPStatus.BAD_REQUEST, {"error": str(exc)})

        def do_PATCH(self) -> None:
            task_id = self.task_id(urlparse(self.path).path)
            if task_id is None:
                self.send_json(HTTPStatus.NOT_FOUND, {"error": "not found"})
                return
            try:
                task = update_task(database, task_id, self.read_json())
            except (ValueError, json.JSONDecodeError) as exc:
                self.send_json(HTTPStatus.BAD_REQUEST, {"error": str(exc)})
                return
            self.send_json(HTTPStatus.OK, task) if task else self.send_json(
                HTTPStatus.NOT_FOUND, {"error": "task not found"}
            )

        def do_DELETE(self) -> None:
            task_id = self.task_id(urlparse(self.path).path)
            if task_id is None or not delete_task(database, task_id):
                self.send_json(HTTPStatus.NOT_FOUND, {"error": "task not found"})
                return
            self.send_response(HTTPStatus.NO_CONTENT)
            self.end_headers()

        def log_message(self, format: str, *args: object) -> None:
            return

    return ThreadingHTTPServer((host, port), Handler)


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("--version", action="version", version="Task Ledger 1.0")
    parser.add_argument("--database", required=True)
    subcommands = parser.add_subparsers(dest="command", required=True)
    list_parser = subcommands.add_parser("list")
    list_parser.add_argument("--status", choices=("all", "open", "completed"), default="all")
    subcommands.add_parser("export")
    serve_parser = subcommands.add_parser("serve")
    serve_parser.add_argument("--host", default="127.0.0.1")
    serve_parser.add_argument("--port", type=int, default=8080)
    args = parser.parse_args()

    if args.command == "list":
        for task in list_tasks(args.database, args.status):
            marker = "x" if task["completed"] else " "
            print(f"{task['id']}\t[{marker}]\t{task['title']}")
        return 0
    if args.command == "export":
        print(
            json.dumps(
                {"contractVersion": "1", "tasks": list_tasks(args.database)},
                ensure_ascii=False,
                indent=2,
            )
        )
        return 0

    server = make_server(args.database, args.host, args.port)
    try:
        server.serve_forever()
    except KeyboardInterrupt:
        pass
    finally:
        server.server_close()
    return 0


if __name__ == "__main__":
    raise SystemExit(main())

<!doctype html>
<meta charset="utf-8">
<meta name="viewport" content="width=device-width,initial-scale=1">
<title>Task Ledger</title>
<style>li span { overflow-wrap: anywhere; }</style>
<h1 id="main-heading" tabindex="-1">Task Ledger</h1>
<form id="new-task"><input id="title" required><button>Add task</button></form>
<label>Show <select id="status"><option>all</option><option>open</option><option>completed</option></select></label>
<ul id="tasks"></ul>
<p id="message" role="status"></p>
<script>
const heading = document.querySelector('#main-heading');
heading.focus();
const tasks = document.querySelector('#tasks');
const message = document.querySelector('#message');
async function request(path, options = {}) {
  const response = await fetch(path, {headers: {'Content-Type': 'application/json'}, ...options});
  if (!response.ok && response.status !== 204) throw new Error(await response.text());
  return response.status === 204 ? null : response.json();
}
async function load({focusTaskId = null, focusAction = null} = {}) {
  try {
    const values = await request('/api/tasks?status=' + document.querySelector('#status').value);
    tasks.replaceChildren();
    if (!values.length) message.textContent = 'No tasks yet.'; else message.textContent = '';
    for (const task of values) {
      const item = document.createElement('li');
      item.dataset.taskId = String(task.id);
      const label = document.createElement('span');
      label.textContent = task.title + (task.completed ? ' (completed)' : '');
      const toggle = document.createElement('button');
      toggle.dataset.action = 'toggle';
      toggle.textContent = task.completed ? 'Reopen' : 'Complete';
      toggle.onclick = async () => {
        await request('/api/tasks/' + task.id, {method: 'PATCH', body: JSON.stringify({completed: !task.completed})});
        await load({focusTaskId: task.id, focusAction: 'toggle'});
      };
      const remove = document.createElement('button');
      remove.dataset.action = 'delete';
      remove.textContent = 'Delete';
      remove.onclick = async () => {
        await request('/api/tasks/' + task.id, {method: 'DELETE'});
        await load();
        document.querySelector('#status').focus();
      };
      item.append(label, ' ', toggle, ' ', remove); tasks.append(item);
    }
    if (focusTaskId !== null && focusAction) {
      document.querySelector(
        `#tasks li[data-task-id="${focusTaskId}"] button[data-action="${focusAction}"]`
      )?.focus();
    }
  } catch (error) {
    message.textContent = 'Could not load tasks.';
    throw error;
  }
}
document.querySelector('#new-task').onsubmit = async event => {
  event.preventDefault();
  const input = document.querySelector('#title');
  await request('/api/tasks', {method: 'POST', body: JSON.stringify({title: input.value})});
  input.value = ''; await load();
};
document.querySelector('#status').onchange = () => { load().catch(() => {}); };
load().catch(() => {});
</script>

from __future__ import annotations

import json
import subprocess
import sys
import tempfile
import threading
import unittest
import urllib.error
import urllib.request
from pathlib import Path

from task_ledger.cli import create_task, list_tasks, make_server, update_task


class TaskLedgerTests(unittest.TestCase):
    def setUp(self) -> None:
        self.temporary = tempfile.TemporaryDirectory()
        self.database = str(Path(self.temporary.name) / "tasks.db")

    def tearDown(self) -> None:
        self.temporary.cleanup()

    def test_crud_filter_and_persistence(self) -> None:
        first = create_task(self.database, "write docs")
        self.assertEqual([task["title"] for task in list_tasks(self.database, "open")], ["write docs"])
        update_task(self.database, int(first["id"]), {"title": "write guide", "completed": True})
        self.assertEqual([task["title"] for task in list_tasks(self.database, "completed")], ["write guide"])
        self.assertEqual(list_tasks(self.database), list_tasks(self.database))

    def test_cli_export_uses_selected_database(self) -> None:
        create_task(self.database, "export me")
        result = subprocess.run(
            [sys.executable, "-m", "task_ledger.cli", "--database", self.database, "export"],
            text=True,
            capture_output=True,
            check=True,
        )
        payload = json.loads(result.stdout)
        self.assertEqual(payload["contractVersion"], "1")
        self.assertEqual(payload["tasks"][0]["title"], "export me")

        help_result = subprocess.run(
            [sys.executable, "-m", "task_ledger.cli", "--database", self.database, "--help"],
            text=True,
            capture_output=True,
            check=False,
        )
        self.assertEqual(help_result.returncode, 0, help_result.stderr)
        self.assertIn("export", help_result.stdout)

        version_result = subprocess.run(
            [sys.executable, "-m", "task_ledger.cli", "--database", self.database, "--version"],
            text=True,
            capture_output=True,
            check=False,
        )
        self.assertEqual(version_result.returncode, 0, version_result.stderr)
        self.assertEqual(version_result.stdout.strip(), "Task Ledger 1.0")

        invalid = subprocess.run(
            [
                sys.executable,
                "-m",
                "task_ledger.cli",
                "--database",
                self.database,
                "list",
                "--status",
                "invalid",
            ],
            text=True,
            capture_output=True,
            check=False,
        )
        self.assertEqual(invalid.returncode, 2)
        self.assertIn("invalid choice", invalid.stderr)

    def test_http_api_positive_and_negative_paths(self) -> None:
        server = make_server(self.database, "127.0.0.1", 0)
        thread = threading.Thread(target=server.serve_forever, daemon=True)
        thread.start()
        base = f"http://127.0.0.1:{server.server_port}"

        def request(method: str, path: str, payload: dict | None = None):
            data = None if payload is None else json.dumps(payload).encode()
            headers = {} if payload is None else {"Content-Type": "application/json"}
            return urllib.request.urlopen(
                urllib.request.Request(base + path, data=data, headers=headers, method=method)
            )

        try:
            health = json.load(request("GET", "/healthz"))
            self.assertEqual(health, {"status": "ok"})
            with self.assertRaises(urllib.error.HTTPError) as missing_health:
                request("GET", "/not-a-service-route")
            self.assertEqual(missing_health.exception.code, 404)

            created = json.load(request("POST", "/api/tasks", {"title": "from api"}))
            with self.assertRaises(urllib.error.HTTPError) as invalid_create:
                request("POST", "/api/tasks", {"title": ""})
            self.assertEqual(invalid_create.exception.code, 400)

            open_tasks = json.load(request("GET", "/api/tasks?status=open"))
            self.assertEqual([task["id"] for task in open_tasks], [created["id"]])
            with self.assertRaises(urllib.error.HTTPError) as invalid_filter:
                request("GET", "/api/tasks?status=invalid")
            self.assertEqual(invalid_filter.exception.code, 400)

            fetched = json.load(request("GET", f"/api/tasks/{created['id']}"))
            self.assertEqual(fetched["title"], "from api")
            with self.assertRaises(urllib.error.HTTPError) as missing_get:
                request("GET", "/api/tasks/999999")
            self.assertEqual(missing_get.exception.code, 404)

            updated = json.load(request("PATCH", f"/api/tasks/{created['id']}", {"completed": True}))
            self.assertTrue(updated["completed"])
            with self.assertRaises(urllib.error.HTTPError) as missing_patch:
                request("PATCH", "/api/tasks/999999", {"completed": True})
            self.assertEqual(missing_patch.exception.code, 404)

            deleted = request("DELETE", f"/api/tasks/{created['id']}")
            self.assertEqual(deleted.status, 204)
            deleted.close()
            with self.assertRaises(urllib.error.HTTPError) as missing_delete:
                request("DELETE", f"/api/tasks/{created['id']}")
            self.assertEqual(missing_delete.exception.code, 404)
        finally:
            server.shutdown()
            server.server_close()
            thread.join()



if __name__ == "__main__":
    unittest.main()

cat > scripts/verify.sh <<'SH'
#!/bin/sh
set -eu
python -m unittest discover -s tests -v
SH
chmod +x scripts/verify.sh

python -m task_ledger.cli --database task-ledger.db serve --host 127.0.0.1 --port 8080

"${CHROME_BINARY:-google-chrome}" --version
"${CHROMEWEBDRIVER:-chromedriver}" --version

python -c '
import hashlib
import pathlib
import urllib.request

data = urllib.request.urlopen("https://raw.githubusercontent.com/TakashiSasaki/templates/7e1352a527cdfa6a20ac5df1a81b404b4a6699b3/examples/onboarding/task-ledger/browser_proof.py", timeout=30).read()
expected = "7921d0308850aeefdb71332c5f089bf6a5d2ed1e50bf5f77b5d3d40eda53030b"
actual = hashlib.sha256(data).hexdigest()
if actual != expected:
    raise SystemExit(f"browser proof SHA-256 mismatch: expected {expected}, got {actual}")
destination = pathlib.Path("tests/test_task_ledger_browser.py")
destination.parent.mkdir(parents=True, exist_ok=True)
destination.write_bytes(data)
print(f"Verified browser proof SHA-256: {actual} ({len(data)} bytes)")
'

cat >> scripts/verify.sh <<'SH'
python tests/test_task_ledger_browser.py
SH

./scripts/verify.sh

python scripts/scaffold_webapp_evidence.py > /tmp/webapp-evidence-worklist.json

{
  "commands": [
    {
      "id": "verify-product",
      "command": "./scripts/verify.sh",
      "purpose": "Run Task Ledger product verification."
    }
  ],
  "releaseGates": [
    {
      "id": "product-verification",
      "purpose": "Require the authoritative product verification command.",
      "commandIds": ["verify-product"]
    }
  ]
}

python /absolute/path/to/agent-skills/composition/scripts/run.py \
  --repository /absolute/path/to/task-ledger \
  validate
python .template-composition/validators/validate_implementation_evidence.py \
  . --release-readiness

./scripts/verify.sh

python /absolute/path/to/agent-skills/composition/scripts/run.py \
  --repository /absolute/path/to/task-ledger \
  inspect
python /absolute/path/to/agent-skills/composition/scripts/run.py \
  --repository /absolute/path/to/task-ledger \
  plan --mode update

python /absolute/path/to/agent-skills/composition/scripts/run.py \
  --repository /absolute/path/to/task-ledger \
  apply --mode update

python /absolute/path/to/agent-skills/composition/scripts/run.py \
  --repository /absolute/path/to/task-ledger \
  plan --mode upgrade --config /absolute/path/to/task-ledger/composition.json
python /absolute/path/to/agent-skills/composition/scripts/run.py \
  --repository /absolute/path/to/task-ledger \
  apply --mode upgrade --config /absolute/path/to/task-ledger/composition.json