Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
127 changes: 123 additions & 4 deletions mcp_server.py
Original file line number Diff line number Diff line change
Expand Up @@ -20,7 +20,7 @@
import os
import re
import sqlite3
from datetime import date, datetime
from datetime import date, datetime, timedelta
from difflib import get_close_matches
from pathlib import Path

Expand Down Expand Up @@ -481,11 +481,130 @@ def get_feature_status(issue_key: str | None = None, title: str | None = None) -
@mcp.tool(
annotations={"readOnlyHint": True, "openWorldHint": False},
)
def release_risk_summary(release: str | None = None) -> str:
def get_release_schedule(
product: str | None = None,
version: str | None = None,
milestone: str | None = None,
) -> str:
"""Get release schedule dates for any product.

All parameters are optional and fuzzy-matched (case-insensitive, partial).
Returns matching milestones ordered by date, including past dates.

Args:
product: Product name. Examples: "AcmeProduct", "acme".
version: Version number. "1.0" also matches "1.0.1", "1.0 EA1", etc.
milestone: Milestone type. Examples: "code freeze", "ga", "release", "planning".

Examples:
get_release_schedule(product="acme", version="1.0", milestone="code freeze")
get_release_schedule(version="2.0")
get_release_schedule(milestone="ga")
"""
conn = _get_conn()
today = date.today()

# --- Query release_schedule table ---
sched_conditions: list[str] = []
sched_params: list[str] = []
if product:
sched_conditions.append("release LIKE ?")
sched_params.append(f"%{product}%")
if version:
sched_conditions.append("release LIKE ?")
sched_params.append(f"%{version}%")
if milestone:
sched_conditions.append("task LIKE ?")
sched_params.append(f"%{milestone}%")

sched_where = " AND ".join(sched_conditions) if sched_conditions else "1=1"
sched_rows = conn.execute(
f"""SELECT release, task AS milestone, date_start, date_finish
FROM release_schedule
WHERE {sched_where}
ORDER BY date_start""",
sched_params,
).fetchall()
Comment on lines +520 to +527
Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🟠 Major

Unbounded result sets can degrade tool latency and memory use.

With no filters, both queries become WHERE 1=1 and return full tables without any cap. This tool should enforce a bounded response like other endpoints that use MAX_QUERY_ROWS.

Proposed fix
-    sched_rows = conn.execute(
+    sched_rows = conn.execute(
         f"""SELECT release, task AS milestone, date_start, date_finish
             FROM release_schedule
             WHERE {sched_where}
-            ORDER BY date_start""",
-        sched_params,
+            ORDER BY date_start
+            LIMIT ?""",
+        [*sched_params, MAX_QUERY_ROWS],
     ).fetchall()
@@
-    mile_rows = conn.execute(
+    mile_rows = conn.execute(
         f"""SELECT product, version, event_type AS milestone, event_date
             FROM release_milestone
             WHERE {mile_where}
-            ORDER BY event_date""",
-        mile_params,
+            ORDER BY event_date
+            LIMIT ?""",
+        [*mile_params, MAX_QUERY_ROWS],
     ).fetchall()

As per coding guidelines, "Focus on major issues impacting performance, readability, maintainability and security. Avoid nitpicks and avoid verbosity."

Also applies to: 542-549

Comment on lines +522 to +527
Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🟠 Major

Date ordering is currently incorrect for mixed date formats.

ORDER BY date_start/event_date sorts raw TEXT, but this file already acknowledges non-ISO date formats via _parse_date(). That can return misordered results while the tool promises date ordering.

Proposed fix
-    sched_rows = conn.execute(
-        f"""SELECT release, task AS milestone, date_start, date_finish
-            FROM release_schedule
-            WHERE {sched_where}
-            ORDER BY date_start""",
-        sched_params,
-    ).fetchall()
+    sched_rows = conn.execute(
+        f"""SELECT release, task AS milestone, date_start, date_finish
+            FROM release_schedule
+            WHERE {sched_where}""",
+        sched_params,
+    ).fetchall()
@@
-    mile_rows = conn.execute(
-        f"""SELECT product, version, event_type AS milestone, event_date
-            FROM release_milestone
-            WHERE {mile_where}
-            ORDER BY event_date""",
-        mile_params,
-    ).fetchall()
+    mile_rows = conn.execute(
+        f"""SELECT product, version, event_type AS milestone, event_date
+            FROM release_milestone
+            WHERE {mile_where}""",
+        mile_params,
+    ).fetchall()
@@
     for entry in schedule:
         d = entry.get("date_finish") or entry.get("date_start")
+        parsed = _parse_date(d) if d else None
+        entry["_sort_date"] = parsed
-        if d:
-            parsed = _parse_date(d)
-            if parsed:
-                entry["status"] = "past" if parsed < today else "upcoming"
-                entry["days_away"] = (parsed - today).days
+        if parsed:
+            entry["status"] = "past" if parsed < today else "upcoming"
+            entry["days_away"] = (parsed - today).days
@@
     for entry in milestones_list:
         d = entry.get("event_date")
-        if d:
-            parsed = _parse_date(d)
-            if parsed:
-                entry["status"] = "past" if parsed < today else "upcoming"
-                entry["days_away"] = (parsed - today).days
+        parsed = _parse_date(d) if d else None
+        entry["_sort_date"] = parsed
+        if parsed:
+            entry["status"] = "past" if parsed < today else "upcoming"
+            entry["days_away"] = (parsed - today).days
+
+    schedule.sort(key=lambda e: e.get("_sort_date") or date.max)
+    milestones_list.sort(key=lambda e: e.get("_sort_date") or date.max)
+    for e in schedule:
+        e.pop("_sort_date", None)
+    for e in milestones_list:
+        e.pop("_sort_date", None)

As per coding guidelines, "Focus on major issues impacting performance, readability, maintainability and security. Avoid nitpicks and avoid verbosity."

Also applies to: 543-549, 554-570

🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.

In `@mcp_server.py` around lines 522 - 527, The SQL ORDER BY on raw TEXT
(date_start/event_date) yields incorrect ordering for mixed date formats; after
fetching rows from the query that returns columns like release, task AS
milestone, date_start, date_finish (and similar queries at the other noted
blocks), remove the ORDER BY clause and instead sort the fetched list in Python
using the existing _parse_date() helper (e.g. rows.sort(key=lambda r:
_parse_date(r.get('date_start') or r.get('event_date')) or datetime.min)) so
dates are normalized and correctly ordered; apply the same change to the other
query sites (the blocks around the referenced lines) and ensure you handle
None/parse failures consistently.


# --- Query release_milestone table ---
mile_conditions: list[str] = []
mile_params: list[str] = []
if product:
mile_conditions.append("product LIKE ?")
mile_params.append(f"%{product}%")
if version:
mile_conditions.append("version LIKE ?")
mile_params.append(f"%{version}%")
if milestone:
mile_conditions.append("event_type LIKE ?")
mile_params.append(f"%{milestone}%")

mile_where = " AND ".join(mile_conditions) if mile_conditions else "1=1"
mile_rows = conn.execute(
f"""SELECT product, version, event_type AS milestone, event_date
FROM release_milestone
WHERE {mile_where}
ORDER BY event_date""",
mile_params,
).fetchall()

schedule = _rows_to_dicts(sched_rows)
milestones_list = _rows_to_dicts(mile_rows)

# Annotate schedule entries with past/future
for entry in schedule:
d = entry.get("date_finish") or entry.get("date_start")
if d:
parsed = _parse_date(d)
if parsed:
entry["status"] = "past" if parsed < today else "upcoming"
entry["days_away"] = (parsed - today).days

# Annotate milestone entries with past/future
for entry in milestones_list:
d = entry.get("event_date")
if d:
parsed = _parse_date(d)
if parsed:
entry["status"] = "past" if parsed < today else "upcoming"
entry["days_away"] = (parsed - today).days

if not schedule and not milestones_list:
hints = []
releases = conn.execute("SELECT DISTINCT release FROM release_schedule ORDER BY release").fetchall()
if releases:
hints = [r["release"] for r in releases]
return json.dumps(
{
"error": "No matching releases found",
"available_releases": hints,
}
)

return json.dumps(
{
"schedule": schedule,
"milestones": milestones_list,
"schedule_count": len(schedule),
"milestone_count": len(milestones_list),
"as_of": today.isoformat(),
},
default=str,
)


@mcp.tool(
annotations={"readOnlyHint": True, "openWorldHint": False},
)
def release_risk_summary(release: str | None = None, lookback_days: int = 30) -> str:
"""Assess release risk by comparing milestone dates against feature completion.

If no release specified, analyzes all releases with upcoming milestones.
If no release specified, analyzes all releases with recent or upcoming milestones.
Flags features under 80% complete when milestone is within 30 days.

Args:
release: Filter to a specific release (fuzzy match on version). Omit for all.
lookback_days: Include milestones up to this many days in the past (default 30).
"""
today = date.today()
conn = _get_conn()
Expand All @@ -504,7 +623,7 @@ def release_risk_summary(release: str | None = None) -> str:
releases_info = {}
for m in milestones:
parsed = _parse_date(m["event_date"], today.year)
if not parsed or parsed < today:
if not parsed or parsed < today - timedelta(days=lookback_days):
continue
key = f"{m['product']} {m['version']}"
if key not in releases_info:
Expand Down
35 changes: 35 additions & 0 deletions scripts/test.sh
Original file line number Diff line number Diff line change
Expand Up @@ -73,6 +73,41 @@ sys.exit(0 if count > 0 else 1)
"
done

# 5b. Tool smoke tests
echo ""
echo "--- Tools ---"
run "get_release_schedule(all)" uv run python3 -c "
import json, sys
sys.path.insert(0, '$REPO_ROOT')
import mcp_server
r = json.loads(mcp_server.get_release_schedule())
assert r.get('schedule_count', 0) > 0, 'no schedule rows'
"
Comment on lines +79 to +85
Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

⚠️ Potential issue | 🟠 Major

Smoke tests are too strict: they ignore milestone-only matches.

Both tests require schedule_count > 0, but get_release_schedule() is valid when only milestone_count is non-zero. This can create false CI failures.

Proposed fix
-r = json.loads(mcp_server.get_release_schedule())
-assert r.get('schedule_count', 0) > 0, 'no schedule rows'
+r = json.loads(mcp_server.get_release_schedule())
+assert (r.get('schedule_count', 0) + r.get('milestone_count', 0)) > 0, 'no release rows'
@@
-r = json.loads(mcp_server.get_release_schedule(product='Acme', version='1.0', milestone='freeze'))
-assert r.get('schedule_count', 0) > 0, 'no filtered rows'
+r = json.loads(mcp_server.get_release_schedule(product='Acme', version='1.0', milestone='freeze'))
+assert (r.get('schedule_count', 0) + r.get('milestone_count', 0)) > 0, 'no filtered rows'

As per coding guidelines, "Focus on major issues impacting performance, readability, maintainability and security. Avoid nitpicks and avoid verbosity."

Also applies to: 87-93

🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.

In `@scripts/test.sh` around lines 79 - 85, The smoke test assertion in the Python
one-liner that calls get_release_schedule() is too strict by requiring
r.get('schedule_count', 0) > 0; change it to accept either schedule rows or
milestone-only results by asserting (r.get('schedule_count', 0) > 0) or
(r.get('milestone_count', 0) > 0). Update both occurrences that run the Python
check (the block using mcp_server.get_release_schedule()) so the new assertion
checks schedule_count OR milestone_count and keep the same error message or
adjust it to reflect "no schedule or milestone rows".


run "get_release_schedule(filtered)" uv run python3 -c "
import json, sys
sys.path.insert(0, '$REPO_ROOT')
import mcp_server
r = json.loads(mcp_server.get_release_schedule(product='Acme', version='1.0', milestone='freeze'))
assert r.get('schedule_count', 0) > 0, 'no filtered rows'
"

run "get_release_schedule(no match)" uv run python3 -c "
import json, sys
sys.path.insert(0, '$REPO_ROOT')
import mcp_server
r = json.loads(mcp_server.get_release_schedule(product='zzz_no_match'))
assert 'error' in r, 'expected error for no match'
"

run "release_risk_summary" uv run python3 -c "
import json, sys
sys.path.insert(0, '$REPO_ROOT')
import mcp_server
r = json.loads(mcp_server.release_risk_summary())
assert 'releases' in r or 'message' in r, 'unexpected response shape'
"

# 6. Schema diff
echo ""
echo "--- Schema ---"
Expand Down
Loading