laily/.pouch
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(ack): add Feishu bug review approval gate

Browse Source
This commit is contained in:
laily
2026-08-03 15:26:02 +08:00
parent 4d078e8258
commit 0954ad542a
15 changed files with 851 additions and 100 deletions
Download Patch File Download Diff File Expand all files Collapse all files
skills/ack/scripts/approval_payload.py
+50
View File
@@ -0,0 +1,50 @@
"""Canonical reviewed Bug task payload shared by intake and validation."""
from __future__ import annotations
import hashlib
import json
import re
from typing import Any
PAYLOAD_FIELDS = (
"title",
"description",
"priority",
"actual",
"expected",
"stepsToReproduce",
"fixLogic",
"acceptanceCriteria",
)
NUMBERED_ITEM = re.compile(r"(?:^|\s)([1-9][0-9]*)\.\s+")
def review_items(value: str) -> list[str]:
"""Recover line or numbered-list review text as stable non-empty items."""
lines = [line.strip(" \t-*•") for line in value.splitlines() if line.strip()]
if len(lines) > 1:
return lines
text = value.strip()
matches = list(NUMBERED_ITEM.finditer(text))
if matches:
items: list[str] = []
for index, match in enumerate(matches):
start = match.end()
end = matches[index + 1].start() if index + 1 < len(matches) else len(text)
item = text[start:end].strip()
if item:
items.append(item)
if items:
return items
return [text] if text else []
def approval_payload_hash(task: dict[str, Any]) -> str:
"""Hash the exact reviewed fields that Developer and Test consume."""
payload = {field: task.get(field) for field in PAYLOAD_FIELDS}
encoded = json.dumps(
payload, ensure_ascii=False, sort_keys=True, separators=(",", ":"),
).encode("utf-8")
return f"sha256:{hashlib.sha256(encoded).hexdigest()}"
skills/ack/scripts/feishu_bug_intake.py
+270 -18
View File
@@ -1,8 +1,9 @@
#!/usr/bin/env python3
"""Read an ACK-ready Feishu Base view through the official lark-cli.
"""Read and review an ACK-ready Feishu Base view through the official lark-cli.
This is deliberately a small, non-mutating adapter. It never reads the
active profile and emits one JSON document only on success.
The only mutation is a bounded draft write to configured Coordinator fields.
The adapter never reads the active profile and emits one JSON document only
on success.
"""
from __future__ import annotations
@@ -23,12 +24,14 @@ import time
from pathlib import Path
from typing import Any
from approval_payload import approval_payload_hash, review_items
from yaml_subset import DuplicateKeyError, YamlSubsetError, load_json_unique, load_yaml_subset, make_unique_pyyaml_loader
REQUIRED_FIELDS = ("title", "actual", "expected", "stepsToReproduce", "acceptance", "priority", "attachments", "updatedAt")
OPTIONAL_FIELDS = ("fixLogic",)
SOURCE_FACT_FIELDS = ("title", "actual", "expected", "updatedAt")
COORDINATOR_FIELDS = ("steps", "acceptance", "priority")
BUG_CONTENT_FIELDS = ("title", "actual", "expected", *COORDINATOR_FIELDS)
BUG_CONTENT_FIELDS = ("title", "actual", "expected", "fixLogic", *COORDINATOR_FIELDS)
MAX_PAGES = 100
MAX_RECORDS = 10_000
PAGE_SIZE = 100
@@ -39,10 +42,13 @@ MAX_TOTAL_ATTACHMENTS = 100
MAX_ATTACHMENT_BYTES = 20 * 1024 * 1024
MAX_TOTAL_ATTACHMENT_BYTES = 200 * 1024 * 1024
MAX_ATTACHMENT_BATCH_SECONDS = 300
MAX_DRAFT_BYTES = 64 * 1024
SAFE_VALUE = re.compile(r"^[^\s\x00-\x1f]{1,256}$")
PROFILE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{0,63}$")
RECORD_ID = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{0,255}$")
SOURCE_REF = re.compile(r"^feishu-base:sha256:[0-9a-f]{64}$")
DRAFT_REVISION = re.compile(r"^sha256:[0-9a-f]{64}$")
WORKFLOWS = {"read-only-v1", "reviewed-writeback-v1"}
class IntakeError(Exception):
@@ -182,6 +188,60 @@ def load_board(path: Path) -> dict[str, Any]:
return value
def load_draft(path: Path) -> dict[str, Any]:
"""Load one bounded, regular JSON file with the two writable draft fields."""
descriptor: int | None = None
try:
before = path.lstat()
if not stat.S_ISREG(before.st_mode) or path.is_symlink():
raise IntakeError("draft input must be a regular file")
flags = os.O_RDONLY | getattr(os, "O_CLOEXEC", 0) | getattr(os, "O_NOFOLLOW", 0)
descriptor = os.open(path, flags)
metadata = os.fstat(descriptor)
if (
not stat.S_ISREG(metadata.st_mode)
or (before.st_dev, before.st_ino) != (metadata.st_dev, metadata.st_ino)
):
raise IntakeError("draft input changed while opening")
if metadata.st_size <= 0 or metadata.st_size > MAX_DRAFT_BYTES:
raise IntakeError("draft input size is invalid")
chunks: list[bytes] = []
total = 0
while total <= MAX_DRAFT_BYTES:
chunk = os.read(descriptor, min(64 * 1024, MAX_DRAFT_BYTES + 1 - total))
if not chunk:
break
chunks.append(chunk)
total += len(chunk)
content = b"".join(chunks)
if len(content) != metadata.st_size:
raise IntakeError("draft input changed while reading")
value = load_json_unique(content.decode("utf-8"))
except IntakeError:
raise
except (OSError, UnicodeError, json.JSONDecodeError, DuplicateKeyError) as exc:
raise IntakeError("cannot read draft input") from exc
finally:
if descriptor is not None:
os.close(descriptor)
if not isinstance(value, dict) or set(value) != {"fixLogic", "acceptance"}:
raise IntakeError("draft input must contain exactly fixLogic and acceptance")
fix_logic = value["fixLogic"]
acceptance = value["acceptance"]
if not isinstance(fix_logic, str) or not fix_logic.strip():
raise IntakeError("draft fixLogic must be a non-empty string")
if (
not isinstance(acceptance, list)
or not acceptance
or any(not isinstance(item, str) or not item.strip() for item in acceptance)
):
raise IntakeError("draft acceptance must be a non-empty string list")
return {
"fixLogic": fix_logic.strip(),
"acceptance": [item.strip() for item in acceptance],
}
def config_from_board(board: dict[str, Any]) -> dict[str, Any]:
project = board.get("project")
if not isinstance(project, dict) or "bugIntake" not in project:
@@ -189,12 +249,15 @@ def config_from_board(board: dict[str, Any]) -> dict[str, Any]:
config = project["bugIntake"]
if not isinstance(config, dict):
raise IntakeError("project.bugIntake must be an object")
allowed = {"provider", "profile", "baseToken", "tableId", "viewId", "fields"}
allowed = {"provider", "workflow", "profile", "baseToken", "tableId", "viewId", "fields"}
unknown = sorted(set(config) - allowed)
if unknown:
raise IntakeError("bugIntake has unknown fields")
if config.get("provider") != "feishu-base":
raise IntakeError("bugIntake.provider must be feishu-base")
workflow = config.get("workflow", "read-only-v1")
if workflow not in WORKFLOWS:
raise IntakeError("bugIntake.workflow is invalid")
profile = config.get("profile")
if not isinstance(profile, str) or not PROFILE.fullmatch(profile):
raise IntakeError("bugIntake.profile is invalid")
@@ -203,12 +266,19 @@ def config_from_board(board: dict[str, Any]) -> dict[str, Any]:
if not isinstance(value, str) or not SAFE_VALUE.fullmatch(value):
raise IntakeError(f"bugIntake.{key} is invalid")
fields = config.get("fields")
if not isinstance(fields, dict) or set(fields) != set(REQUIRED_FIELDS):
raise IntakeError("bugIntake.fields must map exactly the required logical fields")
supported = set(REQUIRED_FIELDS) | set(OPTIONAL_FIELDS)
if (
not isinstance(fields, dict)
or not set(REQUIRED_FIELDS).issubset(fields)
or not set(fields).issubset(supported)
):
raise IntakeError("bugIntake.fields must map all required and only supported logical fields")
if any(not isinstance(value, str) or not SAFE_VALUE.fullmatch(value) for value in fields.values()):
raise IntakeError("bugIntake.fields values are invalid")
if len(set(fields.values())) != len(fields):
raise IntakeError("bugIntake.fields values must be unique")
if workflow == "reviewed-writeback-v1" and "fixLogic" not in fields:
raise IntakeError("reviewed writeback requires bugIntake.fields.fixLogic")
return config
@@ -381,7 +451,11 @@ def matrix_from_response(response: dict[str, Any], field_ids: list[str]) -> tupl
def fetch_pages(config: dict[str, Any]) -> list[tuple[str, list[Any]]]:
field_ids = [config["fields"][logical] for logical in REQUIRED_FIELDS]
logical_fields = [
logical for logical in (*REQUIRED_FIELDS, *OPTIONAL_FIELDS)
if logical in config["fields"]
]
field_ids = [config["fields"][logical] for logical in logical_fields]
all_rows: list[tuple[str, list[Any]]] = []
offset = 0
for _ in range(MAX_PAGES):
@@ -444,6 +518,37 @@ def source_ref(config: dict[str, Any], record_id: str) -> str:
return f"feishu-base:sha256:{hashlib.sha256(identity.encode('utf-8')).hexdigest()}"
def draft_revision(
record: dict[str, Any], attachment_tokens: list[str] | None = None,
) -> str:
"""Bind approval to the normalized source facts and review-controlled fields."""
tokens = attachment_tokens or []
if len(tokens) != len(record["attachments"]):
raise IntakeError("draft revision attachment identity is incomplete")
stable = {
"sourceRef": record["sourceRef"],
"updatedAt": record["updatedAt"],
"title": record["title"],
"actual": record["actual"],
"expected": record["expected"],
"steps": record["steps"],
"fixLogic": record["fixLogic"],
"acceptance": record["acceptance"],
"priority": record["priority"],
"attachments": [
{
**{key: attachment.get(key) for key in ("name", "type", "size")},
"tokenDigest": f"sha256:{hashlib.sha256(('ack-feishu-attachment-v1\x1f' + token).encode('utf-8')).hexdigest()}",
}
for attachment, token in zip(record["attachments"], tokens)
],
}
encoded = json.dumps(
stable, ensure_ascii=False, sort_keys=True, separators=(",", ":"),
).encode("utf-8")
return f"sha256:{hashlib.sha256(encoded).hexdigest()}"
def fetch(config: dict[str, Any], output_dir: Path | None) -> dict[str, Any]:
profile_check(config)
prepared: list[tuple[dict[str, Any], list[tuple[dict[str, Any], str]]]] = []
@@ -451,7 +556,11 @@ def fetch(config: dict[str, Any], output_dir: Path | None) -> dict[str, Any]:
total_attachments = 0
total_attachment_bytes = 0
for record_id, row in fetch_pages(config):
cells = dict(zip(REQUIRED_FIELDS, row))
logical_fields = [
logical for logical in (*REQUIRED_FIELDS, *OPTIONAL_FIELDS)
if logical in config["fields"]
]
cells = dict(zip(logical_fields, row))
attachment_data = attachment_items(cells["attachments"])
total_attachments += len(attachment_data)
total_attachment_bytes += sum(metadata["size"] for metadata, _ in attachment_data)
@@ -459,16 +568,22 @@ def fetch(config: dict[str, Any], output_dir: Path | None) -> dict[str, Any]:
raise IntakeError("batch exceeded the attachment count limit")
if total_attachment_bytes > MAX_TOTAL_ATTACHMENT_BYTES:
raise IntakeError("batch exceeded the attachment byte limit")
record = {"sourceRef": source_ref(config, record_id), "recordId": record_id, "updatedAt": text(cells["updatedAt"]), "title": text(cells["title"]), "actual": text(cells["actual"]), "expected": text(cells["expected"]), "steps": text(cells["stepsToReproduce"]), "acceptance": text(cells["acceptance"]), "priority": text(cells["priority"]), "attachments": [metadata for metadata, _ in attachment_data], "warnings": []}
record = {"sourceRef": source_ref(config, record_id), "recordId": record_id, "updatedAt": text(cells["updatedAt"]), "title": text(cells["title"]), "actual": text(cells["actual"]), "expected": text(cells["expected"]), "steps": text(cells["stepsToReproduce"]), "fixLogic": text(cells.get("fixLogic")), "acceptance": text(cells["acceptance"]), "priority": text(cells["priority"]), "attachments": [metadata for metadata, _ in attachment_data], "warnings": []}
if not attachment_data and not any(record[field] for field in BUG_CONTENT_FIELDS):
batch_warnings.append({"recordId": record_id, "code": "blank_record_skipped"})
continue
for field in SOURCE_FACT_FIELDS:
if not record[field]:
raise IntakeError(f"record {field} must not be empty")
enrichment_fields = list(COORDINATOR_FIELDS)
if "fixLogic" in config["fields"]:
enrichment_fields.append("fixLogic")
record["enrichmentRequired"] = [
field for field in COORDINATOR_FIELDS if not record[field]
field for field in enrichment_fields if not record[field]
]
record["draftRevision"] = draft_revision(
record, [token for _, token in attachment_data],
)
prepared.append((record, attachment_data))
download_root: Path | None = None
if output_dir is not None:
@@ -493,7 +608,117 @@ def fetch(config: dict[str, Any], output_dir: Path | None) -> dict[str, Any]:
attachment["size"], min(60, remaining),
)
records.append(record)
return {"provider": "feishu-base", "profile": config["profile"], "tableId": config["tableId"], "viewId": config["viewId"], "records": records, "warnings": batch_warnings}
return {"provider": "feishu-base", "workflow": config.get("workflow", "read-only-v1"), "profile": config["profile"], "tableId": config["tableId"], "viewId": config["viewId"], "records": records, "warnings": batch_warnings}
def review_record(
config: dict[str, Any], record_id: str, expected_source_ref: str,
expected_revision: str,
) -> dict[str, Any]:
"""Resolve one record inside the configured view and bind its reviewed version."""
if RECORD_ID.fullmatch(record_id) is None:
raise IntakeError("review record id is invalid")
if SOURCE_REF.fullmatch(expected_source_ref) is None:
raise IntakeError("expected source reference is invalid")
if DRAFT_REVISION.fullmatch(expected_revision) is None:
raise IntakeError("expected draft revision is invalid")
matching = [
record for record in fetch(config, None)["records"]
if record["recordId"] == record_id
]
if len(matching) != 1:
raise IntakeError("configured review view did not contain exactly one record")
if (
matching[0]["sourceRef"] != expected_source_ref
or matching[0]["draftRevision"] != expected_revision
):
raise IntakeError("review record changed before the requested operation")
return matching[0]
def write_draft(
config: dict[str, Any], record_id: str, expected_source_ref: str,
expected_revision: str, draft_path: Path,
) -> dict[str, Any]:
"""Overwrite only the configured fix logic and acceptance cells."""
if config.get("workflow", "read-only-v1") != "reviewed-writeback-v1":
raise IntakeError("draft writeback requires reviewed-writeback-v1 workflow")
if "fixLogic" not in config["fields"]:
raise IntakeError("bugIntake.fields.fixLogic is required for draft writeback")
review_record(config, record_id, expected_source_ref, expected_revision)
draft = load_draft(draft_path)
patch = {
config["fields"]["fixLogic"]: draft["fixLogic"],
config["fields"]["acceptance"]: "\n".join(
f"{index}. {item}" for index, item in enumerate(draft["acceptance"], start=1)
),
}
profile_check(config)
run_cli([
"base", "+record-upsert", "--profile", config["profile"],
"--base-token", config["baseToken"], "--table-id", config["tableId"],
"--record-id", record_id, "--json",
json.dumps(patch, ensure_ascii=False, separators=(",", ":")),
"--format", "json",
])
matching = [record for record in fetch(config, None)["records"] if record["recordId"] == record_id]
if len(matching) != 1 or matching[0]["sourceRef"] != expected_source_ref:
raise IntakeError("draft writeback readback did not find exactly one record")
expected_acceptance = text(patch[config["fields"]["acceptance"]])
if (
matching[0]["fixLogic"] != text(draft["fixLogic"])
or matching[0]["acceptance"] != expected_acceptance
):
raise IntakeError("draft writeback readback did not match the submitted draft")
return {
"provider": "feishu-base",
"recordId": record_id,
"written": ["fixLogic", "acceptance"],
"draftRevision": matching[0]["draftRevision"],
"ok": True,
}
def import_approved(
config: dict[str, Any], record_id: str, expected_source_ref: str,
expected_revision: str,
) -> dict[str, Any]:
"""Emit the canonical task payload for one explicitly approved draft revision."""
if config.get("workflow", "read-only-v1") != "reviewed-writeback-v1":
raise IntakeError("approved import requires reviewed-writeback-v1 workflow")
record = review_record(
config, record_id, expected_source_ref, expected_revision,
)
steps = review_items(record["steps"])
acceptance = review_items(record["acceptance"])
if not record["priority"] or not steps or not record["fixLogic"] or not acceptance:
raise IntakeError("approved record is missing prepared review fields")
task_draft: dict[str, Any] = {
"title": record["title"],
"priority": record["priority"],
"description": record["title"],
"actual": record["actual"],
"expected": record["expected"],
"stepsToReproduce": steps,
"fixLogic": record["fixLogic"],
"acceptanceCriteria": acceptance,
"source": {
"kind": "feishu-base",
"workflow": "reviewed-writeback-v1",
"ref": record["sourceRef"],
"recordId": record["recordId"],
"updatedAt": record["updatedAt"],
"approvedRevision": record["draftRevision"],
},
}
task_draft["source"]["approvedPayloadHash"] = approval_payload_hash(task_draft)
return {
"provider": "feishu-base",
"recordId": record_id,
"draftRevision": record["draftRevision"],
"taskDraft": task_draft,
"ok": True,
}
def plan_actions(board: dict[str, Any], records: list[dict[str, Any]]) -> list[dict[str, Any]]:
@@ -529,10 +754,12 @@ def plan_actions(board: dict[str, Any], records: list[dict[str, Any]]) -> list[d
ref = record.get("sourceRef")
record_id = record.get("recordId")
updated_at = record.get("updatedAt")
revision = record.get("draftRevision")
if (
not isinstance(ref, str) or SOURCE_REF.fullmatch(ref) is None
or not isinstance(record_id, str) or RECORD_ID.fullmatch(record_id) is None
or not isinstance(updated_at, str) or not updated_at
or not isinstance(revision, str) or DRAFT_REVISION.fullmatch(revision) is None
):
raise IntakeError("normalized Feishu record identity is invalid")
if ref in seen_records:
@@ -540,14 +767,22 @@ def plan_actions(board: dict[str, Any], records: list[dict[str, Any]]) -> list[d
seen_records.add(ref)
task = existing.get(ref)
if task is None:
planned_action: dict[str, Any] = {"sourceRef": ref, "recordId": record_id, "action": "create"}
planned_action: dict[str, Any] = {"sourceRef": ref, "recordId": record_id, "draftRevision": record["draftRevision"], "action": "create"}
enrichment_required = record.get("enrichmentRequired")
if enrichment_required:
planned_action["enrichmentRequired"] = enrichment_required
actions.append(planned_action)
continue
source = task["source"]
if source["updatedAt"] == updated_at:
approved_revision = source.get("approvedRevision")
if isinstance(approved_revision, str):
if approved_revision == revision:
action_name = "unchanged"
elif task["status"] == "open":
action_name = "refresh"
else:
action_name = "drift"
elif source["updatedAt"] == updated_at:
action_name = "unchanged"
elif task["status"] == "open":
action_name = "refresh"
@@ -556,6 +791,7 @@ def plan_actions(board: dict[str, Any], records: list[dict[str, Any]]) -> list[d
planned_action = {
"sourceRef": ref,
"recordId": record_id,
"draftRevision": record["draftRevision"],
"taskId": task["id"],
"status": task["status"],
"action": action_name,
@@ -568,25 +804,41 @@ def plan_actions(board: dict[str, Any], records: list[dict[str, Any]]) -> list[d
def main(argv: list[str] | None = None) -> int:
parser = argparse.ArgumentParser(description="Read a configured Feishu Base bug intake")
parser = argparse.ArgumentParser(description="Read and review a configured Feishu Base bug intake")
sub = parser.add_subparsers(dest="command", required=True)
for name in ("check", "fetch", "plan"):
for name in ("check", "fetch", "plan", "write-draft", "import-approved"):
command = sub.add_parser(name)
command.add_argument("tasks", type=Path, help="ACK tasks.yaml or JSON board")
if name in {"fetch", "plan"}:
command.add_argument("--output-dir", type=Path, help="explicit directory for downloaded attachments")
if name in {"write-draft", "import-approved"}:
command.add_argument("--record-id", required=True, help="existing Feishu Base record id")
command.add_argument("--expected-source-ref", required=True, help="sourceRef returned by fetch")
command.add_argument("--expected-draft-revision", required=True, help="draftRevision returned by fetch")
if name == "write-draft":
command.add_argument("--input", type=Path, required=True, help="bounded JSON draft file")
args = parser.parse_args(argv)
try:
board = load_board(args.tasks)
config = config_from_board(board)
if args.command == "check":
profile_check(config)
output = {"provider": "feishu-base", "profile": config["profile"], "ok": True}
output = {"provider": "feishu-base", "workflow": config.get("workflow", "read-only-v1"), "profile": config["profile"], "ok": True}
elif args.command == "fetch":
output = fetch(config, args.output_dir)
else:
elif args.command == "plan":
output = fetch(config, args.output_dir)
output["actions"] = plan_actions(board, output["records"])
elif args.command == "write-draft":
output = write_draft(
config, args.record_id, args.expected_source_ref,
args.expected_draft_revision, args.input,
)
else:
output = import_approved(
config, args.record_id, args.expected_source_ref,
args.expected_draft_revision,
)
except IntakeError as exc:
sys.stderr.write(f"Feishu bug intake failed: {exc}\n")
return 1
skills/ack/scripts/validate_tasks.py
+69 -4
View File
@@ -24,6 +24,7 @@ import re
import sys
from pathlib import Path
from approval_payload import approval_payload_hash
from yaml_subset import (
DuplicateKeyError,
YamlSubsetError,
@@ -117,8 +118,13 @@ FEISHU_REQUIRED_FIELDS = {
"title", "actual", "expected", "stepsToReproduce", "acceptance", "priority",
"attachments", "updatedAt",
}
FEISHU_CONFIG_FIELDS = {"provider", "profile", "baseToken", "tableId", "viewId", "fields"}
FEISHU_SOURCE_FIELDS = {"kind", "ref", "recordId", "updatedAt"}
FEISHU_OPTIONAL_FIELDS = {"fixLogic"}
FEISHU_CONFIG_FIELDS = {"provider", "workflow", "profile", "baseToken", "tableId", "viewId", "fields"}
FEISHU_SOURCE_FIELDS = {
"kind", "workflow", "ref", "recordId", "updatedAt", "approvedRevision",
"approvedPayloadHash",
}
FEISHU_WORKFLOWS = {"read-only-v1", "reviewed-writeback-v1"}
FEISHU_PROFILE_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{0,63}$")
FEISHU_SOURCE_REF_RE = re.compile(r"^feishu-base:sha256:[0-9a-f]{64}$")
FEISHU_RECORD_ID_RE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{0,255}$")
@@ -598,6 +604,7 @@ def validate_with_schema(data: dict, schema_path: Path) -> list[str]:
def validate_builtin(data: dict) -> list[str]:
errors: list[str] = []
project_intake_workflow = "read-only-v1"
def validate_string_fields(
value: dict,
@@ -671,6 +678,11 @@ def validate_builtin(data: dict) -> list[str]:
reject_unknown_fields(intake, FEISHU_CONFIG_FIELDS, "project.bugIntake", errors)
if intake.get("provider") != "feishu-base":
errors.append("project.bugIntake.provider 必须是 feishu-base")
workflow = intake.get("workflow", "read-only-v1")
if workflow in FEISHU_WORKFLOWS:
project_intake_workflow = workflow
if workflow not in FEISHU_WORKFLOWS:
errors.append("project.bugIntake.workflow 非法")
profile = intake.get("profile")
if not isinstance(profile, str) or FEISHU_PROFILE_RE.fullmatch(profile) is None:
errors.append("project.bugIntake.profile 非法")
@@ -679,12 +691,18 @@ def validate_builtin(data: dict) -> list[str]:
if not isinstance(value, str) or not value.strip() or any(char.isspace() for char in value):
errors.append(f"project.bugIntake.{key} 必须是无空白非空字符串")
fields = intake.get("fields")
if not isinstance(fields, dict) or set(fields) != FEISHU_REQUIRED_FIELDS:
if (
not isinstance(fields, dict)
or not FEISHU_REQUIRED_FIELDS.issubset(fields)
or not set(fields).issubset(FEISHU_REQUIRED_FIELDS | FEISHU_OPTIONAL_FIELDS)
):
errors.append("project.bugIntake.fields 必须且只能映射所需逻辑字段")
elif any(not isinstance(v, str) or not v.strip() or any(c.isspace() for c in v) for v in fields.values()):
errors.append("project.bugIntake.fields 字段值必须是无空白非空字符串")
elif len(set(fields.values())) != len(fields):
errors.append("project.bugIntake.fields 字段值不能重复")
elif workflow == "reviewed-writeback-v1" and "fixLogic" not in fields:
errors.append("reviewed-writeback-v1 必须映射 project.bugIntake.fields.fixLogic")
if (
"knowledgeFile" in project
and project.get("knowledgeFile") != "docs/ack/knowledge.yaml"
@@ -776,6 +794,7 @@ def validate_builtin(data: dict) -> list[str]:
"assignee",
"component",
"description",
"fixLogic",
"expected",
"actual",
},
@@ -783,7 +802,7 @@ def validate_builtin(data: dict) -> list[str]:
)
validate_string_lists(
task,
{"specRefs", "testRefs", "stepsToReproduce"},
{"specRefs", "testRefs", "stepsToReproduce", "acceptanceCriteria"},
where,
)
validate_object_fields(task, {"evidence", "verification"}, where)
@@ -806,6 +825,52 @@ def validate_builtin(data: dict) -> list[str]:
errors.append(f"{where}.source.recordId: 必须是合法飞书记录 ID")
if not _nonempty_string(source.get("updatedAt")):
errors.append(f"{where}.source.updatedAt: 必须是非空字符串")
source_workflow = source.get("workflow", "read-only-v1")
if source_workflow not in FEISHU_WORKFLOWS:
errors.append(f"{where}.source.workflow: 非法")
if (
project_intake_workflow == "reviewed-writeback-v1"
and source_workflow != "reviewed-writeback-v1"
and status not in {"verified", "leftover"}
):
errors.append(
f"{where}.source.workflow: reviewed 项目的可执行飞书任务必须先迁移审核"
)
approved_revision = source.get("approvedRevision")
stored_payload_hash = source.get("approvedPayloadHash")
if source_workflow == "reviewed-writeback-v1" and approved_revision is None:
errors.append(f"{where}.source.approvedRevision: reviewed workflow 必填")
elif approved_revision is not None and (
not isinstance(approved_revision, str)
or re.fullmatch(r"sha256:[0-9a-f]{64}", approved_revision) is None
):
errors.append(f"{where}.source.approvedRevision: 必须是 sha256 revision")
if source_workflow == "reviewed-writeback-v1":
if (
not isinstance(stored_payload_hash, str)
or re.fullmatch(r"sha256:[0-9a-f]{64}", stored_payload_hash) is None
):
errors.append(f"{where}.source.approvedPayloadHash: reviewed workflow 必填")
required_strings = ("title", "priority", "actual", "expected", "fixLogic")
for field in required_strings:
if not _nonempty_string(task.get(field)):
errors.append(f"{where}.{field}: reviewed workflow 必须是非空字符串")
for field in ("stepsToReproduce", "acceptanceCriteria"):
items = task.get(field)
if (
not isinstance(items, list)
or not items
or any(not _nonempty_string(item) for item in items)
):
errors.append(f"{where}.{field}: reviewed workflow 必须是非空字符串列表")
if (
isinstance(stored_payload_hash, str)
and re.fullmatch(r"sha256:[0-9a-f]{64}", stored_payload_hash)
and stored_payload_hash != approval_payload_hash(task)
):
errors.append(f"{where}.source.approvedPayloadHash: 与任务审核字段不匹配")
elif stored_payload_hash is not None:
errors.append(f"{where}.source.approvedPayloadHash: 只允许 reviewed workflow")
validate_knowledge_fields(task, where, status, errors)