Add TypeAlias ban rule and tinyagent tree hygiene lint

.pre-commit-config.yaml

@@ -68,3 +68,10 @@ repos:
         language: system
         pass_filenames: false
         always_run: true
+
+      - id: treelint
+        name: treelint (tinyagent tree hygiene)
+        entry: python3 scripts/lint_tinyagent_tree.py
+        language: system
+        pass_filenames: false
+        always_run: true

docs/README.md

@@ -324,6 +324,7 @@ Cross-provider tool-call smoke examples:
 - [Prompt Caching](api/caching.md): Cache breakpoints, cost savings, and provider requirements
 - [OpenAI-Compatible Endpoints](api/openai-compatible-endpoints.md): Using `OpenAICompatModel.base_url` with OpenRouter/OpenAI/Chutes-compatible backends
 - [Usage Semantics](api/usage-semantics.md): Canonical `usage` schema across provider flows
+- [AST Rule: No TypeAlias Shims](ast-grep-no-typealias.md): Structural lint rule banning `TypeAlias` / `TypeAliasType`
 - [Changelog](../CHANGELOG.md): Release history
 
 ## Project Structure

docs/ast-grep-no-typealias.md

@@ -0,0 +1,44 @@
+# AST Rule: No `TypeAlias` / `TypeAliasType` Shims
+
+This repository enforces a strict rule to prevent `TypeAlias`-based shim types in Python.
+
+Rule file:
+
+- `src/rules/ast/rules/no_typealias_python.yml`
+
+## Why
+
+`TypeAlias` and `TypeAliasType` can become an abstraction layer that hides real concrete types.
+For this codebase, we prefer direct, explicit types over alias shims.
+
+## What is banned
+
+The rule flags these symbols anywhere in scoped Python files:
+
+- `TypeAlias`
+- `typing.TypeAlias`
+- `typing_extensions.TypeAlias`
+- `TypeAliasType`
+- `typing_extensions.TypeAliasType`
+
+## Scope
+
+The rule currently scans:
+
+- `tinyagent/**/*.py`
+- `tests/**/*.py`
+- `docs/**/*.py`
+
+## Run locally
+
+From repo root:
+
+```bash
+sg scan --config src/rules/ast/sgconfig.yml --filter no-typealias-python tinyagent tests docs
+sg test --config src/rules/ast/sgconfig.yml
+```
+
+## Notes
+
+- This is a structural lint rule (ast-grep), not a runtime check.
+- If you need shared typing structure, use concrete generic types directly in-place instead of alias shims.

rules/README.md

@@ -2,6 +2,43 @@
 
 AST-grep rules for repository-specific enforcement.
 
+## Python type-shim guard (`TypeAlias` ban)
+
+Rule file:
+
+- `src/rules/ast/rules/no_typealias_python.yml`
+
+Purpose:
+
+- ban `TypeAlias` and `TypeAliasType` shim usage in Python code
+
+Run:
+
+```bash
+sg scan --config src/rules/ast/sgconfig.yml --filter no-typealias-python tinyagent tests docs
+```
+
+See also:
+
+- `docs/ast-grep-no-typealias.md`
+
+## tinyagent tree hygiene rule
+
+Rule script:
+
+- `scripts/lint_tinyagent_tree.py`
+
+Purpose:
+
+- reject `__pycache__` directories under `tinyagent/`
+- reject empty/cache-only directories under `tinyagent/`
+
+Run:
+
+```bash
+python3 scripts/lint_tinyagent_tree.py
+```
+
 ## Harness anti-duck-typing guard
 
 Rule files:

scripts/lint_tinyagent_tree.py

@@ -0,0 +1,77 @@
+#!/usr/bin/env python3
+"""Enforce tinyagent package tree hygiene.
+
+Rules:
+- no `__pycache__` directories under `tinyagent/`
+- no empty (or cache-only) directories under `tinyagent/`
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from pathlib import Path
+
+ROOT = Path("tinyagent")
+
+
+@dataclass(frozen=True)
+class Violation:
+    path: Path
+    code: str
+    message: str
+
+
+def _iter_dirs(root: Path) -> list[Path]:
+    return sorted(path for path in root.rglob("*") if path.is_dir())
+
+
+def _has_non_cache_children(directory: Path) -> bool:
+    return any(child.name != "__pycache__" for child in directory.iterdir())
+
+
+def check(root: Path = ROOT) -> list[Violation]:
+    if not root.exists() or not root.is_dir():
+        return [Violation(path=root, code="TREE000", message="tinyagent/ directory not found")]
+
+    violations: list[Violation] = []
+    for directory in _iter_dirs(root):
+        if directory.name == "__pycache__":
+            violations.append(
+                Violation(
+                    path=directory,
+                    code="TREE001",
+                    message="Remove __pycache__ directory from tinyagent/",
+                )
+            )
+            continue
+
+        if not _has_non_cache_children(directory):
+            violations.append(
+                Violation(
+                    path=directory,
+                    code="TREE002",
+                    message="Remove empty/cache-only directory from tinyagent/",
+                )
+            )
+
+    return violations
+
+
+def main() -> int:
+    violations = check()
+    if not violations:
+        print("treelint: tinyagent tree is clean")
+        return 0
+
+    for violation in violations:
+        print(f"{violation.path}  {violation.code}  {violation.message}")
+
+    print("\nCleanup:")
+    print("  find tinyagent -type d -name '__pycache__' -prune -exec rm -rf {} +")
+    print("  find tinyagent -depth -type d -empty ! -path 'tinyagent' -delete")
+    print(f"\ntreelint: {len(violations)} violation(s) found")
+    return 1
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

src/rules/ast/README.md

@@ -5,6 +5,7 @@ ast-grep rules for source code enforcement.
 Current rules:
 
 - `rules/no_any_python.yml`: absolute ban on `Any` in Python type annotations and imports.
+- `rules/no_typealias_python.yml`: bans `TypeAlias` and `TypeAliasType` shims in Python.
 
 Run from repository root:
 

src/rules/ast/rule-tests/__snapshots__/no-typealias-python-snapshot.yml

@@ -0,0 +1,27 @@
+id: no-typealias-python
+snapshots:
+  ? |
+    from typing import TypeAlias
+  : labels:
+    - source: TypeAlias
+      style: primary
+      start: 19
+      end: 28
+  ? |
+    from typing_extensions import TypeAliasType
+
+    JsonValue = TypeAliasType("JsonValue", "str | int")
+  : labels:
+    - source: TypeAliasType
+      style: primary
+      start: 30
+      end: 43
+  ? |
+    import typing
+
+    JsonObject: typing.TypeAlias = dict[str, object]
+  : labels:
+    - source: typing.TypeAlias
+      style: primary
+      start: 27
+      end: 43

src/rules/ast/rule-tests/no-typealias-python-test.yml

@@ -0,0 +1,20 @@
+id: no-typealias-python
+valid:
+  - |
+    from typing import Callable
+
+    Handler = Callable[[int], str]
+  - |
+    def label(value: int) -> str:
+        return str(value)
+invalid:
+  - |
+    from typing import TypeAlias
+  - |
+    import typing
+
+    JsonObject: typing.TypeAlias = dict[str, object]
+  - |
+    from typing_extensions import TypeAliasType
+
+    JsonValue = TypeAliasType("JsonValue", "str | int")

src/rules/ast/rules/no_typealias_python.yml

@@ -0,0 +1,15 @@
+id: no-typealias-python
+language: python
+severity: error
+message: "Do not use TypeAlias/TypeAliasType shims. Use concrete types directly."
+files:
+  - tinyagent/**/*.py
+  - tests/**/*.py
+  - docs/**/*.py
+rule:
+  any:
+    - pattern: TypeAlias
+    - pattern: typing.TypeAlias
+    - pattern: typing_extensions.TypeAlias
+    - pattern: TypeAliasType
+    - pattern: typing_extensions.TypeAliasType

tinyagent/proxy.py

@@ -115,18 +115,10 @@ def _build_proxy_request_body(
     }
 
 
-async def _parse_error_response(response: httpx.Response) -> str:
-    """Parse error message from a non-200 proxy response."""
+def _build_proxy_error_message(response: httpx.Response) -> str:
+    """Build a deterministic error message for non-200 proxy responses."""
 
-    error_message = f"Proxy error: {response.status_code}"
-    try:
-        error_data = await response.aread()
-        error_json = json.loads(error_data)
-        if isinstance(error_json, dict) and error_json.get("error"):
-            error_message = f"Proxy error: {error_json['error']}"
-    except Exception:
-        pass
-    return error_message
+    return f"Proxy error: {response.status_code}"
 
 
 def _parse_sse_lines(buffer: str, chunk: str) -> tuple[str, list[str]]:
@@ -164,6 +156,13 @@ async def _iter_sse_events(response: httpx.Response) -> AsyncIterator[JsonObject
                 yield data
 
 
+class _QueueDoneSignal:
+    """Sentinel that marks the end of the proxy event stream."""
+
+
+_QUEUE_DONE = _QueueDoneSignal()
+
+
 class ProxyStreamResponse(StreamResponse):
     """A streaming response that reads SSE from a proxy server."""
 
@@ -175,17 +174,17 @@ def __init__(self, *, model: Model, context: Context, options: ProxyStreamOption
         self._partial: AssistantMessage = _create_initial_partial(model)
         self._final: AssistantMessage | None = None
 
-        self._queue: asyncio.Queue[AssistantMessageEvent | None] = asyncio.Queue()
+        self._queue: asyncio.Queue[AssistantMessageEvent | _QueueDoneSignal] = asyncio.Queue()
         self._task = asyncio.create_task(self._run())
 
     def __aiter__(self) -> AsyncIterator[AssistantMessageEvent]:
         return self
 
     async def __anext__(self) -> AssistantMessageEvent:
-        event = await self._queue.get()
-        if event is None:
+        queued_item = await self._queue.get()
+        if isinstance(queued_item, _QueueDoneSignal):
             raise StopAsyncIteration
-        return event
+        return queued_item
 
     async def result(self) -> AssistantMessage:
         await self._task
@@ -240,7 +239,7 @@ async def _run_success(self) -> None:
                 timeout=None,
             ) as response:
                 if response.status_code != 200:
-                    raise RuntimeError(await _parse_error_response(response))
+                    raise RuntimeError(_build_proxy_error_message(response))
 
                 await self._stream_from_http_response(response)
 
@@ -267,7 +266,7 @@ async def _run(self) -> None:
         except Exception as exc:  # noqa: BLE001
             await self._run_error(exc)
         finally:
-            await self._queue.put(None)
+            await self._queue.put(_QUEUE_DONE)
 
 
 async def stream_proxy(