Files
infra-plugins/plugins/taiga-auto-sync/hooks/taiga-write-guard.py
Adnan Zahir 6a9c18f5a5 fix(taiga-auto-sync): match real Taiga MCP tool names and drop impossible owner/sprint checks
The write-guard's matcher regex never matched the actual tool names
(createUserStory/createTask/createIssue, not user_story_create etc.), so
the metadata guard has never actually fired. It also checked for
assigned_to/milestone fields that don't exist on any create tool's
schema, which would have denied every create once the matcher was fixed.
Guard now validates only the tag contract (area/source/target), covers
the batch-create tools per item, and SKILL.md documents the real
follow-up calls (assignIssue/assignUserStoryToSprint/addIssueToSprint)
needed to set owner and sprint after creation.
2026-09-14 00:31:53 +07:00

97 lines
3.2 KiB
Python

#!/usr/bin/env python3
"""Reject Taiga item creation that is missing the required tag contract.
Only checks tags. Owner and sprint are not creation-time fields on the
Taiga MCP server's create tools (createUserStory/createTask/createIssue and
their batch variants) -- they are set afterwards via assignIssue,
assignUserStoryToSprint, or addIssueToSprint. Tasks have no assignee/sprint
tool at all, so their owner and sprint are tracked through the parent
User Story instead.
"""
import json
import sys
AREAS = {
"infra", "network", "cloud", "security", "backup", "monitoring",
"access", "automation", "support", "vendor",
}
# Batch-create tools nest each item's tags under one of these list keys
# instead of exposing a top-level "tags" array.
BATCH_LIST_KEYS = ("userStories", "tasks", "issues")
def missing_tag_fields(tags):
tags = set(tags or [])
missing = []
if not tags.intersection(AREAS):
missing.append("area tag")
if not any(tag.startswith("source:") for tag in tags):
missing.append("source tag")
if not any(tag.startswith(("repo:", "system:")) for tag in tags):
missing.append("target tag")
return missing
def missing_fields(payload):
item = payload.get("tool_input", {})
for key in BATCH_LIST_KEYS:
entries = item.get(key)
if isinstance(entries, list):
problems = []
for index, entry in enumerate(entries):
missing = missing_tag_fields(entry.get("tags"))
if missing:
label = entry.get("subject") or f"item {index}"
problems.append(f"{label!r}: missing " + ", ".join(missing))
return problems
return missing_tag_fields(item.get("tags"))
def result(payload):
missing = missing_fields(payload)
if not missing:
return None
return {
"hookSpecificOutput": {
"hookEventName": "PreToolUse",
"permissionDecision": "deny",
"permissionDecisionReason": (
"Taiga item is missing required tags: " + "; ".join(missing) + ". "
"Apply one area tag, one source:<agent> tag, and one "
"repo:<name>/system:<name> tag. Owner and sprint are not "
"creation-time fields here -- set them right after creation "
"with assignIssue / assignUserStoryToSprint / addIssueToSprint "
"(Tasks have no assignee/sprint tool; track Task ownership and "
"sprint through the parent User Story)."
),
}
}
def main():
if sys.argv[1:] == ["--self-check"]:
ok_tags = ["infra", "source:codex", "repo:brain"]
assert result({"tool_input": {"tags": ok_tags}}) is None
assert result({"tool_input": {"tags": []}}) is not None
ok_batch = {"tool_input": {"issues": [{"subject": "a", "tags": ok_tags}]}}
assert result(ok_batch) is None
bad_batch = {"tool_input": {"issues": [
{"subject": "a", "tags": ok_tags},
{"subject": "b", "tags": []},
]}}
assert result(bad_batch) is not None
print("self-check: ok")
return
output = result(json.load(sys.stdin))
if output:
print(json.dumps(output))
if __name__ == "__main__":
main()