taolib.symphony.tracker

taolib.symphony.tracker#

Tracker 子包 — Issue Tracker 客户端抽象与实现。

提供统一的 TrackerClient 接口和 Linear 具体实现， 供编排层通过依赖反转访问外部 issue 跟踪器。

Submodules#

Exceptions#

LinearAPIRequestError

网络 / 传输层错误。

LinearAPIStatusError

非 2xx 响应状态码。

LinearError

所有 Linear API 相关错误的基类。

LinearGraphQLError

GraphQL 响应中 errors 字段非空。

LinearMissingCursorError

分页响应缺少 endCursor。

Classes#

TrackerClient

Issue Tracker 客户端抽象基类。

LinearClient

Linear API 客户端，基于 gql + httpx 异步传输。

Issue

规范化后的 Issue 领域模型。

TrackerConfig

Tracker 客户端配置。

Package Contents#

class taolib.symphony.tracker.TrackerClient#

Bases: abc.ABC

Issue Tracker 客户端抽象基类。

子类需实现以下三个查询方法，供编排层在轮询周期中调用：

abstractmethod fetch_candidate_issues() → list[taolib.symphony.tracker.models.Issue]#
Async:

获取候选 issue 列表。

返回当前活跃状态下、属于指定项目的所有 issue， 用于编排层的调度决策。

abstractmethod fetch_issues_by_states(states: list[str]) → list[taolib.symphony.tracker.models.Issue]#
Async:

按状态集合过滤 issue。

主要用于终态清理（如启动时扫描 Done / Cancelled 状态）。

参数:

states -- 状态名称列表（大小写敏感，与 tracker 原始值一致）。

abstractmethod fetch_issue_states_by_ids(issue_ids: list[str]) → list[taolib.symphony.tracker.models.Issue]#
Async:

根据 ID 列表批量获取 issue 的当前状态。

用于协调循环（reconciliation）中刷新运行中 issue 的最新状态， 以检测是否已被外部转为终态。

参数:

issue_ids -- Issue UUID 列表。

exception taolib.symphony.tracker.LinearAPIRequestError#

Bases: LinearError

网络 / 传输层错误。

在 HTTP 请求本身无法发出或连接中断时抛出， 例如 DNS 解析失败、连接超时、TLS 握手错误等。

exception taolib.symphony.tracker.LinearAPIStatusError(message: str, *, status_code: int)#

Bases: LinearError

非 2xx 响应状态码。

携带 status_code 属性以方便调用方区分 4xx / 5xx。

status_code#
exception taolib.symphony.tracker.LinearError#

Bases: taolib.symphony.errors.TrackerError

所有 Linear API 相关错误的基类。

便于调用方一次性捕获全部 Linear 异常，也可按子类精细处理。

exception taolib.symphony.tracker.LinearGraphQLError(message: str, *, errors: list[dict])#

Bases: LinearError

GraphQL 响应中 errors 字段非空。

Linear API 在业务层校验失败时返回此类错误， 例如字段不存在、权限不足、变量类型不匹配等。

errors#
exception taolib.symphony.tracker.LinearMissingCursorError#

Bases: LinearError

分页响应缺少 endCursor。

当 Linear 返回 hasNextPage=True 却未提供 endCursor 时抛出，表示分页完整性受损，无法继续翻页。

class taolib.symphony.tracker.LinearClient(config: taolib.symphony.tracker.models.TrackerConfig)#

Bases: taolib.symphony.tracker.base.TrackerClient

Linear API 客户端，基于 gql + httpx 异步传输。

参数:

config -- Tracker 配置，包含 endpoint、api_key、project_slug 等。

_client#
_config#
async fetch_candidate_issues() → list[taolib.symphony.tracker.models.Issue]#

获取候选 issue 列表（分页）。

按 project slug + 配置中的 active_states 过滤， 每页取 _PAGE_SIZE 条，自动翻页直到遍历完成。

async fetch_issues_by_states(states: list[str]) → list[taolib.symphony.tracker.models.Issue]#

按指定状态集合过滤 issue（分页）。

主要用于终态清理：启动时扫描 Done / Cancelled 等终态 的工作区残留。

参数:

states -- 状态名称列表。

async fetch_issue_states_by_ids(issue_ids: list[str]) → list[taolib.symphony.tracker.models.Issue]#

根据 ID 列表批量获取 issue 的当前状态（分批）。

将 ID 列表按 _PAGE_SIZE 分批查询，每批独立发起请求， 最终按原始请求顺序排列返回结果。

参数:

issue_ids -- Issue UUID 列表。

async _paginated_query(query: object, variables: dict, field_name: str) → list[dict]#

通用分页查询逻辑。

自动跟踪 endCursor，循环翻页直到 hasNextPage 为 False。

参数:

  • query -- gql 编译后的 AST 查询文档。

  • variables -- GraphQL 变量字典。

  • field_name -- 响应 data 中对应的字段名（如 "issues"）。

返回:

所有页的 node 字典合并列表。

抛出:

LinearMissingCursorError -- hasNextPage=True 但 endCursor 缺失。

async _execute_query(query: object, variables: dict, field_name: str, *, paginated: bool) → tuple[list[dict], dict]#

执行单次 GraphQL 查询并提取结果。

参数:

  • query -- gql AST 查询文档。

  • variables -- GraphQL 变量。

  • field_name -- 响应 data 中的字段名。

  • paginated -- 是否为分页查询（需要提取 pageInfo）。

返回:

(nodes, page_info) 元组。 非分页查询的 page_info 为空字典。

抛出:
class taolib.symphony.tracker.Issue#

Bases: pydantic.BaseModel

规范化后的 Issue 领域模型。

由 Linear 等外部 tracker 返回的原始数据经 normalize 后得到统一结构， 供编排层、模板渲染和可观测性输出使用。

id: str#
identifier: str#
title: str#
description: str | None = None#
priority: int | None#
url: str | None = None#
state: str#
labels: list[str]#
blocked_by: list[str]#
branch_name: str | None = None#
created_at: datetime.datetime | None = None#
updated_at: datetime.datetime | None = None#
class taolib.symphony.tracker.TrackerConfig#

Bases: pydantic.BaseModel

Tracker 客户端配置。

从 WORKFLOW.md front matter 的 tracker 段解析而来， 也可由环境变量覆盖。

kind: str = 'linear'#

Tracker 类型，当前仅支持 linear。

api_key: str#

Linear API Key，通常以 lin_api_ 开头。

endpoint: str = 'https://api.linear.app/graphql'#

GraphQL 端点地址。

project_slug: str#

项目 slug，用于候选 issue 的过滤条件。

active_states: list[str]#
timeout: float = 30.0#

HTTP 请求超时（秒）。

max_retries: int = 3#

传输层最大重试次数。