feat(workday): add identity drift mismatch scanners and MCP tool coverage (#2)

* expanded the Workday fixture data in data.py so each mock worker now carries the paired source attributes needed for upcoming drift checks: Workday title, AD title, Workday cost center context, AD department, legal name, preferred name, AD display name, and AD enabled state. Existing fields used by the current tools were preserved, so this was a schema expansion rather than a behavioral change.

* added a dedicated data-layer detector in data.py named scan_status_reconciliation_mismatches(). It scans the current mock worker set and flags the exact mismatch you asked for: status == "Terminated" in Workday while ad_enabled == True in AD. The response shape follows the existing manager-scan pattern with scan_summary plus a mismatches list.

* exposed the status reconciliation detector as a dedicated MCP tool in server.py by importing the new data-layer function and adding scan_status_reconciliation(). This keeps the existing architecture intact: the server remains a thin wrapper, and the scan logic stays in data.py.

* added scan_job_title_mismatches() to data.py. It compares workday_title against ad_title for each worker and returns the same structured pattern as the other detectors: scan_summary plus a mismatches list with both source values, a deterministic mismatch type, and severity.

* exposed the title detector as a dedicated MCP tool in server.py by importing the new data-layer function and adding scan_job_title_drift(). The server is still acting as a thin wrapper over the detector in data.py.

* added scan_department_drift() to data.py. It compares the Workday department context against the AD department string and includes the Workday cost center in each mismatch record so the drift is easier to interpret during reconciliation.

* exposed the department detector as a dedicated MCP tool in server.py by adding scan_department_mismatches(), which delegates to the detector in data.py. The existing tool structure remains unchanged.

* added scan_name_variance() to data.py. The detector normalizes legal, preferred, and AD display names, allows reasonable preferred-name usage, and flags cases where the AD display name no longer aligns to either the legal or preferred first name plus the legal last name.

* exposed the name-variance detector as a dedicated MCP tool in server.py by adding scan_name_variance_mismatches(), which delegates to the detector in data.py. The server still remains a thin wrapper layer.

* I’m checking the Workday project’s current test layout first so I can add one focused test file that matches the existing environment and runner configuration.

Searched for files matching `**/Workday/workday-mcp/**/*`, 3 matches

The Workday runtime is even thinner than the planning docs imply: there’s no test package or local pyproject in that folder yet. I’m checking the parent Workday artifacts and the root repo config so the new tests land in a runnable location.

Read [](file:///c%3A/Users/castn1.CORP/OneDrive%20-%20Wheels/Repos/mcp_servers/Workday/workday-mcp)

Searched for files matching `**/pyproject.toml`, 1 match

Searched for files matching `**/pytest.ini`, no matches

Searched for files matching `**/tox.ini`, no matches

I’ve confirmed there’s no existing Workday test harness, so I’m borrowing the minimal testing conventions from the Identity project before I add one focused detector test file.

Searched for regex `pytest|testpaths|pythonpath` (`**/Identity/pyproject.toml`), 2 results

Read [](file:///c%3A/Users/castn1.CORP/OneDrive%20-%20Wheels/Repos/mcp_servers/Identity/pyproject.toml)

Fetching Python environment information
Completed with input: {
  "resourcePath": "c:\\Users\\castn1.CORP\\OneDrive - Wheels\\Repos\\mcp_servers\\Workday\\workday-mcp"
}

I have enough context to add the test file. The Workday virtual environment does not currently include `pytest`, so I’m installing the minimal test dependency now and then I’ll add a single focused test module for the new detectors and wrappers.

Installing pytest>=7.4.0 in Python 3.13.12 (.venv)
Completed with input: {
  "resourcePath": "c:\\Users\\castn1.CORP\\OneDrive - Wheels\\Repos\\mcp_servers\\Workday\\workday-mcp",
  "packageList": [
    "pytest>=7.4.0"
  ]
}

Completed: *Expose status MCP tool* (11/18)

The test dependency is installed. I’m adding one focused Workday test module now for the four detectors and their MCP wrapper functions.

Generating patch (95 lines) in [](file:///c%3A/Users/castn1.CORP/OneDrive%20-%20Wheels/Repos/mcp_servers/Workday/workday-mcp/tests/test_mismatch_scans.py)

```

```
The test file is in place. I’m running only that focused test module now and checking for any file errors introduced by the new coverage.

Ran terminal command:  cd "/c/Users/castn1.CORP/OneDrive - Wheels/Repos/mcp_servers/Workday/workday-mcp" && "c:/Users/castn1.CORP/OneDrive - Wheels/Repos/mcp_servers/Workday/workday-mcp/.venv/Scripts/python.exe" -m pytest tests/test_mismatch_scans.py -q

Checked [](file:///c%3A/Users/castn1.CORP/OneDrive%20-%20Wheels/Repos/mcp_servers/Workday/workday-mcp/tests/test_mismatch_scans.py), [](file:///c%3A/Users/castn1.CORP/OneDrive%20-%20Wheels/Repos/mcp_servers/Workday/workday-mcp/server.py), [](file:///c%3A/Users/castn1.CORP/OneDrive%20-%20Wheels/Repos/mcp_servers/Workday/workday-mcp/lib/data.py), no problems found

[[ PROCESSING: Mode=NORMAL_MODE | Intent=Report completion of the focused test-coverage step and pause before any optional documentation/status update | Strategy=Summarize the new test file and the verification result, then wait for NEXT ]]

Step 10 is complete on `feat/workday-mismatch-expansion`.

I added focused pytest coverage in test_mismatch_scans.py for the four new detectors and their MCP wrappers:
`scan_status_reconciliation_mismatches`, `scan_job_title_mismatches`, `scan_department_drift`, `scan_name_variance`, plus the corresponding server functions in server.py.

Verification passed. Running the focused test module succeeded with `8 passed in 0.49s`, and there are no reported errors in test_mismatch_scans.py, server.py, or data.py.

Reply exactly with:

NEXT

After that, I’ll do the final atomic step: update status or planning documentation only where needed to reflect the newly implemented mismatch coverage, then I’ll close the session with a summary, snapshot draft, and a Conventional Commit suggestion for your branch review flow.

Made changes.

* Updated to reflect changes

---------

Co-authored-by: nathan <nathan@castaldifamily.com>
This commit is contained in:
Nathan Castaldi 2026-04-03 13:39:53 -04:00 committed by GitHub
parent 36a34876d7
commit 042ce34837
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
4 changed files with 365 additions and 3 deletions

View File

@ -30,6 +30,7 @@ This page is the high-visibility execution status for Workday Integration Suite
| Memory-backed worker status tool (`get_worker_status`) | WIS-009 | 🟢 Green | Returns structured allowlist-oriented worker payloads from deterministic fixtures. |
| Manager lookup tool (`get_worker_manager`) | WIS-011, WIS-017 prep | 🟢 Green | Resolves valid, missing-manager, and unresolved-manager scenarios. |
| Manager mismatch detector (`scan_manager_mismatches`) | WIS-017 | 🟢 Green | Functional prototype scans full mock set and reports unresolved manager links. |
| Expanded identity drift detectors (`scan_status_reconciliation`, `scan_job_title_drift`, `scan_department_mismatches`, `scan_name_variance_mismatches`) | WIS-014 to WIS-018 scope | 🟢 Green | Mock-backed drift scans now cover terminated-but-enabled, title drift, department drift, and legal-vs-display name review scenarios with focused pytest coverage. |
| API token flow and real Workday backend | WIS-008 | 🟡 Yellow | Design path is clear; implementation still pending non-prod credentials and auth closure. |
| Non-prod auth/access unblockers | WIS-001 to WIS-003 | 🔴 Red | External decisions and access provisioning remain gate conditions for API-mode validation. |
@ -48,7 +49,7 @@ This page is the high-visibility execution status for Workday Integration Suite
| --- | --- | --- | --- |
| Unblockers and access readiness | WIS-001 to WIS-005 | 🔴 Red | Pre-implementation dependencies are defined but not yet closed. |
| Core Workday MCP buildout | WIS-006 to WIS-012 | 🟡 Yellow | Modular and memory-backed foundation is live; API/auth and resilience controls are next. |
| Correlation and mismatch expansion | WIS-013 to WIS-018 | 🟡 Yellow | WIS-017 manager mismatch path is prototyped; remaining mismatch categories are queued. |
| Correlation and mismatch expansion | WIS-013 to WIS-018 | 🟡 Yellow | Multiple mock-backed mismatch detectors are now implemented and tested; live directory/API correlation and remaining roadmap categories are still pending. |
| Automation, reporting, remediation | WIS-019 to WIS-030 | 🔴 Red | Flow automation, KPI instrumentation, and cutover remain roadmap backlog. |
## Recent activity (from git history)
@ -57,6 +58,8 @@ This page is the high-visibility execution status for Workday Integration Suite
- Added formalized README update prompt for repeatable status refreshes.
- Refined Workday runtime modular structure and validated three core tools.
- Completed type-hint quality refinements consistent with Pylance discipline.
- Added four mismatch-detection tools for status, title, department, and name variance review.
- Added focused pytest coverage for Workday mismatch scans and MCP wrappers.
## Next milestone focus

View File

@ -4,66 +4,295 @@ from typing import Any
MOCK_WORKERS: dict[str, dict[str, Any]] = {
"EMP001": {
"name": "Nathan",
"legal_name": "Nathaniel Cole",
"preferred_name": "Nathan",
"ad_display_name": "Nathan Cole",
"status": "Active",
"ad_enabled": True,
"dept": "IT",
"workday_cost_center": "CC100-IT",
"workday_title": "Systems Engineer",
"ad_title": "Systems Engineer",
"ad_department": "IT",
"email": "nathan@example.com",
"manager_id": "EMP010",
},
"EMP002": {
"name": "Terminated User",
"legal_name": "Taylor Brooks",
"preferred_name": "Taylor",
"ad_display_name": "Taylor Brooks",
"status": "Terminated",
"ad_enabled": True,
"dept": "Sales",
"workday_cost_center": "CC200-SALES",
"workday_title": "Account Executive",
"ad_title": "Account Executive",
"ad_department": "Sales",
"email": "user2@example.com",
"manager_id": "EMP020",
},
"EMP003": {
"name": "Alicia",
"legal_name": "Alicia Gomez",
"preferred_name": "Alicia",
"ad_display_name": "Alicia Gomez",
"status": "Active",
"ad_enabled": True,
"dept": "IT",
"workday_cost_center": "CC100-IT",
"workday_title": "Senior Systems Analyst",
"ad_title": "Systems Analyst",
"ad_department": "IT",
"email": "alicia@example.com",
"manager_id": "EMP010",
},
"EMP004": {
"name": "Jordan",
"legal_name": "Jordan Lee",
"preferred_name": "Jordan",
"ad_display_name": "Jordan Lee",
"status": "Leave",
"ad_enabled": True,
"dept": "Finance",
"workday_cost_center": "CC300-FIN",
"workday_title": "Finance Analyst",
"ad_title": "Finance Analyst",
"ad_department": "Accounting",
"email": "jordan@example.com",
"manager_id": "EMP030",
},
"EMP010": {
"name": "Priya Manager",
"legal_name": "Priya Narayanan",
"preferred_name": "Priya",
"ad_display_name": "Priya Manager",
"status": "Active",
"ad_enabled": True,
"dept": "IT",
"workday_cost_center": "CC110-IT-MGMT",
"workday_title": "IT Manager",
"ad_title": "IT Manager",
"ad_department": "IT",
"email": "priya@example.com",
"manager_id": "EMP100",
},
"EMP020": {
"name": "Ramon Director",
"legal_name": "Ramon Alvarez",
"preferred_name": "Ramon",
"ad_display_name": "Ramon Director",
"status": "Active",
"ad_enabled": True,
"dept": "Sales",
"workday_cost_center": "CC210-SALES-MGMT",
"workday_title": "Sales Director",
"ad_title": "Sales Director",
"ad_department": "Sales",
"email": "ramon@example.com",
"manager_id": "EMP100",
},
"EMP030": {
"name": "Morgan Lead",
"legal_name": "Morgan Patel",
"preferred_name": "Morgan",
"ad_display_name": "Morgan Patel",
"status": "Active",
"ad_enabled": True,
"dept": "Finance",
"workday_cost_center": "CC310-FIN-MGMT",
"workday_title": "Finance Lead",
"ad_title": "Finance Lead",
"ad_department": "Finance",
"email": "morgan@example.com",
"manager_id": "EMP100",
},
"EMP100": {
"name": "Chief Exec",
"legal_name": "Evelyn Carter",
"preferred_name": "Evelyn",
"ad_display_name": "Evelyn Carter",
"status": "Active",
"ad_enabled": True,
"dept": "Executive",
"workday_cost_center": "CC999-EXEC",
"workday_title": "Chief Executive Officer",
"ad_title": "Chief Executive Officer",
"ad_department": "Executive",
"email": "ceo@example.com",
"manager_id": "",
},
# Intentional unresolved manager reference for mismatch test scenarios
"EMP777": {
"name": "Mismatch Case",
"legal_name": "Alexandra Rivers",
"preferred_name": "Alex",
"ad_display_name": "Jordan Rivers",
"status": "Active",
"ad_enabled": True,
"dept": "Operations",
"workday_cost_center": "CC400-OPS",
"workday_title": "Operations Specialist",
"ad_title": "Operations Specialist",
"ad_department": "Operations",
"email": "mismatch@example.com",
"manager_id": "EMP999",
},
}
def scan_status_reconciliation_mismatches() -> dict[str, Any]:
"""Detect workers terminated in Workday but still enabled in AD."""
mismatches: list[dict[str, Any]] = []
total_scanned = 0
for employee_id, details in MOCK_WORKERS.items():
total_scanned += 1
workday_status = details.get("status")
ad_enabled = bool(details.get("ad_enabled", False))
if workday_status == "Terminated" and ad_enabled:
mismatches.append(
{
"employee_id": employee_id,
"employee_name": details["name"],
"workday_status": workday_status,
"ad_enabled": ad_enabled,
"mismatch_type": "terminated_but_enabled",
"severity": "high",
}
)
return {
"scan_summary": {
"total_records_checked": total_scanned,
"mismatches_found": len(mismatches),
"status": "action_required" if mismatches else "clean",
},
"mismatches": mismatches,
}
def scan_job_title_mismatches() -> dict[str, Any]:
"""Detect workers whose Workday title differs from their AD title."""
mismatches: list[dict[str, Any]] = []
total_scanned = 0
for employee_id, details in MOCK_WORKERS.items():
total_scanned += 1
workday_title = details.get("workday_title", "")
ad_title = details.get("ad_title", "")
if workday_title and ad_title and workday_title != ad_title:
mismatches.append(
{
"employee_id": employee_id,
"employee_name": details["name"],
"workday_title": workday_title,
"ad_title": ad_title,
"mismatch_type": "job_title_mismatch",
"severity": "medium",
}
)
return {
"scan_summary": {
"total_records_checked": total_scanned,
"mismatches_found": len(mismatches),
"status": "action_required" if mismatches else "clean",
},
"mismatches": mismatches,
}
def scan_department_drift() -> dict[str, Any]:
"""Detect workers whose Workday department context differs from AD department."""
mismatches: list[dict[str, Any]] = []
total_scanned = 0
for employee_id, details in MOCK_WORKERS.items():
total_scanned += 1
workday_department = details.get("dept", "")
workday_cost_center = details.get("workday_cost_center", "")
ad_department = details.get("ad_department", "")
if workday_department and ad_department and workday_department != ad_department:
mismatches.append(
{
"employee_id": employee_id,
"employee_name": details["name"],
"workday_department": workday_department,
"workday_cost_center": workday_cost_center,
"ad_department": ad_department,
"mismatch_type": "department_drift",
"severity": "medium",
}
)
return {
"scan_summary": {
"total_records_checked": total_scanned,
"mismatches_found": len(mismatches),
"status": "action_required" if mismatches else "clean",
},
"mismatches": mismatches,
}
def _normalize_name_tokens(value: str) -> list[str]:
return [token for token in value.lower().replace(".", " ").split() if token]
def scan_name_variance() -> dict[str, Any]:
"""Detect AD display names that do not align to legal or preferred Workday names."""
mismatches: list[dict[str, Any]] = []
total_scanned = 0
for employee_id, details in MOCK_WORKERS.items():
total_scanned += 1
legal_name = details.get("legal_name", "")
preferred_name = details.get("preferred_name", "")
ad_display_name = details.get("ad_display_name", "")
if not legal_name or not ad_display_name:
continue
legal_tokens = _normalize_name_tokens(legal_name)
preferred_tokens = _normalize_name_tokens(preferred_name)
display_tokens = _normalize_name_tokens(ad_display_name)
if not legal_tokens or not display_tokens:
continue
legal_first = legal_tokens[0]
legal_last = legal_tokens[-1]
preferred_first = preferred_tokens[0] if preferred_tokens else ""
display_first = display_tokens[0]
display_last = display_tokens[-1]
first_name_aligned = display_first in {legal_first, preferred_first}
last_name_aligned = display_last == legal_last
if first_name_aligned and last_name_aligned:
continue
mismatches.append(
{
"employee_id": employee_id,
"employee_name": details["name"],
"workday_legal_name": legal_name,
"workday_preferred_name": preferred_name,
"ad_display_name": ad_display_name,
"mismatch_type": "name_variance_requires_review",
"severity": "low",
}
)
return {
"scan_summary": {
"total_records_checked": total_scanned,
"mismatches_found": len(mismatches),
"status": "action_required" if mismatches else "clean",
},
"mismatches": mismatches,
}

View File

@ -1,6 +1,12 @@
from mcp.server.fastmcp import FastMCP
from typing import Any
from lib.data import MOCK_WORKERS
from lib.data import (
MOCK_WORKERS,
scan_department_drift,
scan_job_title_mismatches,
scan_name_variance,
scan_status_reconciliation_mismatches,
)
mcp = FastMCP("Workday-Sync") #Server name for MCP registration, e.g. "Workday-Sync"
@ -93,5 +99,37 @@ def scan_manager_mismatches() -> dict[str, Any]:
"mismatches": mismatches
}
@mcp.tool()
def scan_status_reconciliation() -> dict[str, Any]:
"""
WIS-014: Flag workers terminated in Workday but still enabled in AD.
"""
return scan_status_reconciliation_mismatches()
@mcp.tool()
def scan_job_title_drift() -> dict[str, Any]:
"""
Detect workers whose Workday title differs from their AD title.
"""
return scan_job_title_mismatches()
@mcp.tool()
def scan_department_mismatches() -> dict[str, Any]:
"""
Detect workers whose Workday department differs from their AD department.
"""
return scan_department_drift()
@mcp.tool()
def scan_name_variance_mismatches() -> dict[str, Any]:
"""
Detect AD display names that do not align with Workday legal or preferred names.
"""
return scan_name_variance()
if __name__ == "__main__":
mcp.run()

View File

@ -0,0 +1,92 @@
from lib.data import (
scan_department_drift,
scan_job_title_mismatches,
scan_name_variance,
scan_status_reconciliation_mismatches,
)
from server import (
scan_department_mismatches,
scan_job_title_drift,
scan_name_variance_mismatches,
scan_status_reconciliation,
)
def test_scan_status_reconciliation_mismatches_returns_expected_record() -> None:
result = scan_status_reconciliation_mismatches()
assert result["scan_summary"]["total_records_checked"] == 9
assert result["scan_summary"]["mismatches_found"] == 1
assert result["mismatches"] == [
{
"employee_id": "EMP002",
"employee_name": "Terminated User",
"workday_status": "Terminated",
"ad_enabled": True,
"mismatch_type": "terminated_but_enabled",
"severity": "high",
}
]
def test_scan_job_title_mismatches_returns_expected_record() -> None:
result = scan_job_title_mismatches()
assert result["scan_summary"]["total_records_checked"] == 9
assert result["scan_summary"]["mismatches_found"] == 1
assert result["mismatches"] == [
{
"employee_id": "EMP003",
"employee_name": "Alicia",
"workday_title": "Senior Systems Analyst",
"ad_title": "Systems Analyst",
"mismatch_type": "job_title_mismatch",
"severity": "medium",
}
]
def test_scan_department_drift_returns_expected_record() -> None:
result = scan_department_drift()
assert result["scan_summary"]["total_records_checked"] == 9
assert result["scan_summary"]["mismatches_found"] == 1
assert result["mismatches"] == [
{
"employee_id": "EMP004",
"employee_name": "Jordan",
"workday_department": "Finance",
"workday_cost_center": "CC300-FIN",
"ad_department": "Accounting",
"mismatch_type": "department_drift",
"severity": "medium",
}
]
def test_scan_name_variance_returns_expected_records() -> None:
result = scan_name_variance()
assert result["scan_summary"]["total_records_checked"] == 9
assert result["scan_summary"]["mismatches_found"] == 3
assert [item["employee_id"] for item in result["mismatches"]] == [
"EMP010",
"EMP020",
"EMP777",
]
def test_scan_status_reconciliation_tool_matches_detector() -> None:
assert scan_status_reconciliation() == scan_status_reconciliation_mismatches()
def test_scan_job_title_drift_tool_matches_detector() -> None:
assert scan_job_title_drift() == scan_job_title_mismatches()
def test_scan_department_mismatches_tool_matches_detector() -> None:
assert scan_department_mismatches() == scan_department_drift()
def test_scan_name_variance_mismatches_tool_matches_detector() -> None:
assert scan_name_variance_mismatches() == scan_name_variance()