better workflows

baseline_scores.json

@@ -1,8 +1,8 @@
 {
   "scores": {
-    "workflow_A": 0.613,
-    "workflow_B": 0.65,
-    "workflow_C": 0.5
+    "workflow_A": 0.697,
+    "workflow_B": 0.744,
+    "workflow_C": 0.722
   },
-  "average": 0.5877
+  "average": 0.721
 }

inference.py

@@ -76,18 +76,38 @@ CRITICAL RULES:
 1. Read schema_hints FIRST — if "jira.priority" → "severity", use "severity" not "priority" in args.
 2. Complete ALL pending_steps in order.
 3. Do not repeat a successful action.
-4. If an operation fails, read the message carefully and adapt.
-5. Use list_* operations to discover record IDs when needed.
-6. Stop when pending_steps is empty or done=true.
+4. When an operation creates a new resource (e.g. create_issue returns issue_id "JIRA-051"),
+   use THAT returned ID for all subsequent operations on that resource — not a pre-existing ID.
+5. If an operation fails, read the message carefully and adapt.
+6. Use list_* operations to discover record IDs when needed — never assume an ID.
+7. Stop when pending_steps is empty or done=true.
 
 Example actions:
+  # Workflow A (bug fix) — discover then chain ticket_id and returned issue_id:
   {"app": "zendesk", "operation": "list_tickets", "args": {"state": "new"}}
   {"app": "zendesk", "operation": "acknowledge_ticket", "args": {"ticket_number": "<ticket_number from list_tickets>"}}
   {"app": "jira", "operation": "create_issue", "args": {"title": "Bug fix for <customer>", "linked_zendesk": "<ticket_number>"}}
-  {"app": "salesforce", "operation": "list_accounts", "args": {"health": "red"}}
-  {"app": "salesforce", "operation": "get_account", "args": {"account_id": "<account_id from list_accounts>"}}
+  # → create_issue returns {"issue_id": "JIRA-051", ...} — use JIRA-051 for assign_owner below
+  {"app": "jira", "operation": "assign_owner", "args": {"issue_id": "<issue_id from create_issue>", "assignee": "<engineer>"}}
+
+  # Workflow B (onboarding) — find the pending employee, then thread employee_id + territory:
   {"app": "workday", "operation": "list_employees", "args": {"status": "pending"}}
-  {"app": "workday", "operation": "log_sla_event", "args": {"ticket_id": "<ticket_number>", "sla_met": true}}
+  # → returns exactly one record. Capture employee_id (e.g. "EMP-NEW-001") AND territory (e.g. "west").
+  {"app": "workday", "operation": "create_onboarding_task", "args": {"employee_id": "<employee_id from list>"}}
+  {"app": "workday", "operation": "provision_access", "args": {"employee_id": "<employee_id from list>", "app_name": "jira"}}
+  # SF: assign that employee as owner of an account in THEIR territory:
+  {"app": "salesforce", "operation": "list_accounts", "args": {"territory": "<territory from list_employees>"}}
+  {"app": "salesforce", "operation": "assign_account_owner", "args": {"account_id": "<account_id from list>", "owner": "<employee_id from list>"}}
+  # Jira: assign an open issue to the new hire's employee_id:
+  {"app": "jira", "operation": "list_issues", "args": {"status": "open"}}
+  {"app": "jira", "operation": "assign_owner", "args": {"issue_id": "<issue_id from list>", "assignee": "<employee_id from list_employees>"}}
+
+  # Workflow C (churn) — discover the at-risk account, then scope all queries to it:
+  {"app": "salesforce", "operation": "list_accounts", "args": {"health": "red"}}
+  {"app": "salesforce", "operation": "flag_churn_risk", "args": {"account_id": "<account_id from list>"}}
+  {"app": "zendesk", "operation": "get_ticket", "args": {"customer_id": "<account_id from list>"}}
+  {"app": "jira", "operation": "list_issues", "args": {"customer_id": "<account_id from list>"}}
+  {"app": "salesforce", "operation": "assign_account_owner", "args": {"account_id": "<account_id from list>", "owner": "<engineer>"}}
 """
 
 WORKFLOW_NAMES = {

server/apps/jira.py

@@ -19,7 +19,7 @@ class JiraApp(BaseApp):
         # Workflow completion state tracking
         self._linked_issues: set = set()    # issue_ids linked to a Zendesk ticket
         self._assigned_issues: set = set()  # issue_ids with a non-null assignee
-        self._bugs_checked: bool = False    # list_issues was called (Workflow C)
+        self._bugs_checked_for: set = set() # customer_ids queried via list_issues (Workflow C)
 
     # ------------------------------------------------------------------
     # BaseApp interface
@@ -29,7 +29,7 @@ class JiraApp(BaseApp):
         self._records = {r["issue_id"]: r for r in records}
         self._linked_issues.clear()
         self._assigned_issues.clear()
-        self._bugs_checked = False
+        self._bugs_checked_for.clear()
         # Seed state from loaded data
         for issue_id, rec in self._records.items():
             if rec.get("assignee"):
@@ -78,12 +78,33 @@ class JiraApp(BaseApp):
         return len(self._linked_issues) > 0
 
     def issue_assigned(self) -> bool:
-        """True once JIRA-001 (primary bug) has an assignee (Workflow A step A4)."""
-        return bool(self._records.get("JIRA-001", {}).get("assignee"))
-
-    def bugs_checked(self) -> bool:
-        """True once list_issues has been called (Workflow C step C3)."""
-        return self._bugs_checked
+        """True once any linked issue has an assignee (Workflow A step A4).
+        Prefers the primary bug; falls back to any linked+assigned issue."""
+        # Primary path: the primary bug was linked and assigned
+        primary = next(
+            (r for r in self._records.values() if r.get("_is_primary_bug")), None
+        )
+        if primary and primary.get("assignee") and primary["issue_id"] in self._linked_issues:
+            return True
+        # Fallback: any issue that's both linked to Zendesk and has an assignee
+        return any(
+            issue_id in self._linked_issues and self._records[issue_id].get("assignee")
+            for issue_id in self._linked_issues
+        )
+
+    def bugs_checked_for(self, account_id: str) -> bool:
+        """True once list_issues was called WITH customer_id=account_id (Workflow C step C3).
+        Tightened from the old free-pass `bugs_checked` flag — the agent must now scope the
+        query to the at-risk account specifically, forcing real cross-app data flow from C1."""
+        return bool(account_id) and account_id in self._bugs_checked_for
+
+    def new_hire_assigned_to_issue(self, employee_id: str) -> bool:
+        """True once any Jira issue's assignee equals employee_id (Workflow B step B4).
+        Forces the agent to use the new hire's employee_id (discovered in B1) when
+        assigning a Jira issue — real Workday → Jira data flow, no free-pass."""
+        if not employee_id:
+            return False
+        return any(r.get("assignee") == employee_id for r in self._records.values())
 
     # ------------------------------------------------------------------
     # Operations
@@ -229,7 +250,10 @@ class JiraApp(BaseApp):
 
     def _op_list_issues(self, status: str = "open", customer_id: Optional[str] = None,
                         limit: int = 10) -> Dict:
-        self._bugs_checked = True
+        # Track which customer_id was queried — used by bugs_checked_for() (Workflow C C3).
+        # Bare list_issues() with no filter no longer satisfies C3.
+        if customer_id:
+            self._bugs_checked_for.add(customer_id)
         matching = [
             r for r in self._records.values()
             if (status == "all" or r.get("status") == status)
@@ -244,248 +268,4 @@ class JiraApp(BaseApp):
                    for r in drifted]
         return {"success": True, "data": compact,
                 "message": f"Found {len(compact)} {status} issues"
-                           + (f" for {customer_id}" if customer_id else "")}
-# """Jira-like app — engineering ticket management."""
-
-# from typing import Dict, List, Optional
-# from server.apps.base_app import BaseApp
-# from server.schema_drift import SchemaDriftEngine
-
-
-# class JiraApp(BaseApp):
-#     APP_NAME = "jira"
-
-#     OPERATIONS = [
-#         "get_issue", "create_issue", "update_status", "set_priority",
-#         "assign_owner", "add_label", "link_zendesk_ticket", "close_issue", "list_issues",
-#     ]
-
-#     def __init__(self, drift: SchemaDriftEngine):
-#         super().__init__(drift)
-#         self._records: Dict[str, Dict] = {}
-#         # Workflow completion state tracking
-#         self._linked_issues: set = set()    # issue_ids linked to a Zendesk ticket
-#         self._assigned_issues: set = set()  # issue_ids with a non-null assignee
-#         self._bugs_checked: bool = False    # list_issues was called (Workflow C)
-
-#     # ------------------------------------------------------------------
-#     # BaseApp interface
-#     # ------------------------------------------------------------------
-
-#     def initialize(self, records: List[Dict]) -> None:
-#         self._records = {r["issue_id"]: r for r in records}
-#         self._linked_issues.clear()
-#         self._assigned_issues.clear()
-#         self._bugs_checked = False
-#         # Seed state from loaded data
-#         for issue_id, rec in self._records.items():
-#             if rec.get("assignee"):
-#                 self._assigned_issues.add(issue_id)
-#             if rec.get("linked_zendesk"):
-#                 self._linked_issues.add(issue_id)
-
-#     def execute(self, operation: str, args: Dict) -> Dict:
-#         method = getattr(self, f"_op_{operation}", None)
-#         if method is None:
-#             return {
-#                 "success": False,
-#                 "message": f"Unknown operation '{operation}'. Available: {', '.join(self.OPERATIONS)}",
-#             }
-#         try:
-#             return method(**args)
-#         except TypeError as exc:
-#             return {"success": False, "message": f"Bad args for '{operation}': {exc}"}
-
-#     def get_state_view(self, max_rows: int = 5) -> str:
-#         open_issues = [r for r in self._records.values()
-#                        if r.get("status") not in ("closed",)][:max_rows]
-#         if not open_issues:
-#             return "No open issues."
-#         lines = []
-#         for rec in open_issues:
-#             view = self._to_agent_view(rec)
-#             keep = ["issue_id", "title",
-#                     "priority", "severity", "urgency_level",
-#                     "assignee", "owner", "assigned_to",
-#                     "status", "state", "current_state",
-#                     "customer_id", "linked_zendesk"]
-#             compact = {k: v for k, v in view.items() if k in keep and v is not None}
-#             lines.append(str(compact))
-#         return "\n".join(lines)
-
-#     def count_open_items(self) -> int:
-#         return sum(1 for r in self._records.values() if r.get("status") != "closed")
-
-#     # ------------------------------------------------------------------
-#     # Workflow completion state checks
-#     # ------------------------------------------------------------------
-
-#     def has_linked_issue(self) -> bool:
-#         """True once any issue is linked to a Zendesk ticket (Workflow A step A2)."""
-#         return len(self._linked_issues) > 0
-
-#     def issue_assigned(self) -> bool:
-#         """True once JIRA-001 (primary bug) has an assignee (Workflow A step A4)."""
-#         return bool(self._records.get("JIRA-001", {}).get("assignee"))
-
-#     def bugs_checked(self) -> bool:
-#         """True once list_issues has been called (Workflow C step C3)."""
-#         return self._bugs_checked
-
-#     # ------------------------------------------------------------------
-#     # Operations
-#     # ------------------------------------------------------------------
-
-#     def _op_get_issue(self, issue_id: str) -> Dict:
-#         rec = self._records.get(issue_id)
-#         if not rec:
-#             return {"success": False, "message": f"Issue {issue_id} not found. Use list_issues to browse."}
-#         return {"success": True, "data": self._to_agent_view(rec),
-#                 "message": f"Retrieved {issue_id}"}
-
-#     def _op_create_issue(self, title: str, **kwargs) -> Dict:
-#         schema_error, schema_adapted = self._check_schema_drift(kwargs)
-#         if schema_error:
-#             return {
-#                 "success": False,
-#                 "schema_error": schema_error,
-#                 "message": (f"Schema error: field '{schema_error}' is not in the current schema. "
-#                             f"Check schema_hints for the correct field name."),
-#             }
-
-#         issue_id = f"JIRA-{len(self._records) + 1:03d}"
-#         # Accept both canonical and drifted names for priority / assignee
-#         priority = (kwargs.get("priority") or kwargs.get("severity")
-#                     or kwargs.get("urgency_level", "p2"))
-#         linked   = kwargs.get("linked_zendesk") or kwargs.get("zendesk_ticket")
-
-#         rec = {
-#             "issue_id":       issue_id,
-#             "title":          title,
-#             "priority":       priority,
-#             "assignee":       kwargs.get("assignee") or kwargs.get("owner") or kwargs.get("assigned_to"),
-#             "status":         "open",
-#             "reporter":       kwargs.get("reporter", "agent"),
-#             "customer_id":    kwargs.get("customer_id"),
-#             "linked_zendesk": linked,
-#             "labels":         [],
-#             "created_at":     "2026-04-21T09:00:00",
-#         }
-#         self._records[issue_id] = rec
-
-#         if linked:
-#             self._linked_issues.add(issue_id)
-#         if rec["assignee"]:
-#             self._assigned_issues.add(issue_id)
-
-#         return {
-#             "success": True,
-#             "data": {"issue_id": issue_id},
-#             "schema_adapted": schema_adapted,
-#             "message": f"Created {issue_id}: '{title}'"
-#                        + (f" linked to {linked}" if linked else ""),
-#         }
-
-#     def _op_update_status(self, issue_id: str, **kwargs) -> Dict:
-#         schema_error, schema_adapted = self._check_schema_drift(kwargs)
-#         if schema_error:
-#             return {"success": False, "schema_error": schema_error,
-#                     "message": f"Schema error: use current field name, not '{schema_error}'"}
-
-#         rec = self._records.get(issue_id)
-#         if not rec:
-#             return {"success": False, "message": f"Issue {issue_id} not found"}
-
-#         new_status = (kwargs.get("status") or kwargs.get("state")
-#                       or kwargs.get("current_state"))
-#         if not new_status:
-#             return {"success": False, "message": "Provide status/state/current_state value"}
-
-#         rec["status"] = new_status
-#         return {"success": True, "schema_adapted": schema_adapted,
-#                 "message": f"{issue_id} status → '{new_status}'"}
-
-#     def _op_set_priority(self, issue_id: str, **kwargs) -> Dict:
-#         schema_error, schema_adapted = self._check_schema_drift(kwargs)
-#         if schema_error:
-#             return {"success": False, "schema_error": schema_error,
-#                     "message": f"Schema error: '{schema_error}' is a stale field name"}
-
-#         rec = self._records.get(issue_id)
-#         if not rec:
-#             return {"success": False, "message": f"Issue {issue_id} not found"}
-
-#         new_priority = (kwargs.get("priority") or kwargs.get("severity")
-#                         or kwargs.get("urgency_level"))
-#         if not new_priority:
-#             return {"success": False,
-#                     "message": "Provide priority / severity / urgency_level value"}
-
-#         rec["priority"] = new_priority
-#         return {"success": True, "schema_adapted": schema_adapted,
-#                 "message": f"{issue_id} priority → '{new_priority}'"}
-
-#     def _op_assign_owner(self, issue_id: str, **kwargs) -> Dict:
-#         schema_error, schema_adapted = self._check_schema_drift(kwargs)
-#         if schema_error:
-#             hint = self._drift.translate_field("assignee", self.APP_NAME)
-#             return {"success": False, "schema_error": schema_error,
-#                     "message": f"Schema error: use '{hint}' instead of '{schema_error}'"}
-
-#         rec = self._records.get(issue_id)
-#         if not rec:
-#             return {"success": False, "message": f"Issue {issue_id} not found"}
-
-#         assignee = (kwargs.get("assignee") or kwargs.get("owner")
-#                     or kwargs.get("assigned_to"))
-#         if not assignee:
-#             correct_field = self._drift.translate_field("assignee", self.APP_NAME)
-#             return {"success": False,
-#                     "message": f"Missing assignee field. Use '{correct_field}' as the arg key for this episode."}
-
-#         rec["assignee"] = assignee
-#         self._assigned_issues.add(issue_id)
-#         return {"success": True, "schema_adapted": schema_adapted,
-#                 "message": f"{issue_id} assigned to '{assignee}'"}
-
-#     def _op_add_label(self, issue_id: str, label: str) -> Dict:
-#         rec = self._records.get(issue_id)
-#         if not rec:
-#             return {"success": False, "message": f"Issue {issue_id} not found"}
-#         rec.setdefault("labels", []).append(label)
-#         return {"success": True, "message": f"Added label '{label}' to {issue_id}"}
-
-#     def _op_link_zendesk_ticket(self, issue_id: str, zendesk_ticket_number: str) -> Dict:
-#         rec = self._records.get(issue_id)
-#         if not rec:
-#             return {"success": False, "message": f"Issue {issue_id} not found"}
-#         rec["linked_zendesk"] = zendesk_ticket_number
-#         self._linked_issues.add(issue_id)
-#         return {"success": True,
-#                 "message": f"Linked {issue_id} ↔ Zendesk {zendesk_ticket_number}"}
-
-#     def _op_close_issue(self, issue_id: str) -> Dict:
-#         rec = self._records.get(issue_id)
-#         if not rec:
-#             return {"success": False, "message": f"Issue {issue_id} not found"}
-#         rec["status"] = "closed"
-#         return {"success": True, "message": f"Closed {issue_id}"}
-
-#     def _op_list_issues(self, status: str = "open", customer_id: Optional[str] = None,
-#                         limit: int = 10) -> Dict:
-#         self._bugs_checked = True
-#         matching = [
-#             r for r in self._records.values()
-#             if (status == "all" or r.get("status") == status)
-#             and (customer_id is None or r.get("customer_id") == customer_id)
-#         ][:limit]
-#         drifted = [self._to_agent_view(r) for r in matching]
-#         keep = ["issue_id", "title", "priority", "severity", "urgency_level",
-#                 "assignee", "owner", "assigned_to",
-#                 "status", "state", "current_state",
-#                 "customer_id", "linked_zendesk"]
-#         compact = [{k: v for k, v in r.items() if k in keep and v is not None}
-#                    for r in drifted]
-#         return {"success": True, "data": compact,
-#                 "message": f"Found {len(compact)} {status} issues"
-#                            + (f" for {customer_id}" if customer_id else "")}
+                           + (f" for {customer_id}" if customer_id else "")}

server/apps/salesforce.py

@@ -65,20 +65,44 @@ class SalesforceApp(BaseApp):
     # ------------------------------------------------------------------
 
     def account_checked(self) -> bool:
-        """True once get_account was called for ACME-001 (Workflow A step A3)."""
-        return bool(self._records.get("ACME-001", {}).get("_account_checked"))
+        """True once get_account was called for the Workflow A customer (Workflow A step A3)."""
+        return any(
+            r.get("_is_workflow_a_account") and r.get("_account_checked")
+            for r in self._records.values()
+        )
 
     def churn_flagged(self) -> bool:
-        """True once flag_churn_risk was called for ACME-003 (Workflow C step C1)."""
-        return bool(self._records.get("ACME-003", {}).get("_churn_flagged"))
+        """True once flag_churn_risk was called for the at-risk account (Workflow C step C1)."""
+        return any(
+            r.get("_is_churn_target") and r.get("_churn_flagged")
+            for r in self._records.values()
+        )
 
     def team_assigned(self) -> bool:
-        """True once assign_account_owner was called (Workflow B step B3)."""
+        """Legacy free-pass check — kept for backwards compatibility.
+        Workflow B no longer uses this; see new_hire_assigned_in_territory()."""
         return any(r.get("_team_assigned") for r in self._records.values())
 
+    def new_hire_assigned_in_territory(self, employee_id: str, territory: str) -> bool:
+        """True once an SF account in `territory` has `employee_id` as its owner
+        (Workflow B step B3 — tightened from the free-pass team_assigned check).
+        Forces real cross-app data flow: the agent must use the employee_id and territory
+        discovered in B1 to satisfy this check."""
+        if not employee_id or not territory:
+            return False
+        return any(
+            r.get("territory") == territory
+            and r.get("owner") == employee_id
+            and r.get("_team_assigned")
+            for r in self._records.values()
+        )
+
     def intervention_assigned(self) -> bool:
-        """True once assign_account_owner called on ACME-003 (Workflow C step C4)."""
-        return bool(self._records.get("ACME-003", {}).get("_intervention_assigned"))
+        """True once assign_account_owner called on the churn-risk account (Workflow C step C4)."""
+        return any(
+            r.get("_is_churn_target") and r.get("_intervention_assigned")
+            for r in self._records.values()
+        )
 
     # ------------------------------------------------------------------
     # Operations
@@ -168,7 +192,9 @@ class SalesforceApp(BaseApp):
 
             rec["owner"] = new_owner
             rec["_team_assigned"] = True
-            if account_id == "ACME-003":
+            # Semantic-marker driven: any churn target getting an owner is an intervention.
+            # Replaces the old `account_id == "ACME-003"` hardcoded ID check.
+            if rec.get("_is_churn_target"):
                 rec["_intervention_assigned"] = True
 
             return {"success": True, "schema_adapted": schema_adapted,

server/apps/workday.py

@@ -68,16 +68,29 @@ class WorkdayApp(BaseApp):
         return any(r.get("_sla_logged") for r in self._records.values())
 
     def employee_created(self) -> bool:
-        """True once create_onboarding_task was called for EMP-NEW-001 (Workflow B step B1)."""
-        return bool(self._records.get("EMP-NEW-001", {}).get("_onboarding_created"))
+        """True once create_onboarding_task was called for the pending new hire (Workflow B step B1)."""
+        return any(
+            r.get("_is_new_hire") and r.get("_onboarding_created")
+            for r in self._records.values()
+        )
 
     def access_provisioned(self, app_name: str) -> bool:
-        """True once provision_access was called for the given app (Workflow B step B2)."""
+        """True once provision_access was called for the NEW HIRE specifically (Workflow B step B2).
+        Tightened from any-employee free-pass to requiring _is_new_hire — eliminates the
+        old training dead-zone where provisioning a random employee satisfied the check."""
         return any(
-            r.get("_access_provisioned", {}).get(app_name)
+            r.get("_is_new_hire") and r.get("_access_provisioned", {}).get(app_name.lower())
             for r in self._records.values()
         )
 
+    def get_new_hire(self) -> Optional[Dict]:
+        """Return the new-hire employee record (the one with _is_new_hire), or None.
+        Used by workflow_engine.py to thread employee_id / territory into B3 / B4 checks."""
+        return next(
+            (r for r in self._records.values() if r.get("_is_new_hire")),
+            None,
+        )
+
     # ------------------------------------------------------------------
     # Operations
     # ------------------------------------------------------------------
@@ -125,7 +138,7 @@ class WorkdayApp(BaseApp):
         if not rec:
             return {"success": False, "message": f"Employee {employee_id} not found"}
 
-        rec.setdefault("_access_provisioned", {})[app_name] = True
+        rec.setdefault("_access_provisioned", {})[app_name.lower()] = True
         return {"success": True, "schema_adapted": schema_adapted,
                 "message": f"Provisioned {app_name} access for {employee_id} ({rec.get('name', '')})"}
 

server/apps/zendesk.py

@@ -62,8 +62,11 @@ class ZendeskApp(BaseApp):
     # ------------------------------------------------------------------
 
     def ticket_acknowledged(self) -> bool:
-        """True once ZD-001 has been acknowledged (Workflow A step A1)."""
-        return bool(self._records.get("ZD-001", {}).get("_acknowledged"))
+        """True once the primary p1/p0 ticket has been acknowledged (Workflow A step A1)."""
+        return any(
+            r.get("_is_primary_ticket") and r.get("_acknowledged")
+            for r in self._records.values()
+        )
 
     def support_queried(self, account_id: str) -> bool:
         """True once tickets for account_id were listed (Workflow C step C2)."""
@@ -256,262 +259,4 @@ class ZendeskApp(BaseApp):
         }
         self._records[employee_id] = profile
         return {"success": True,
-                "message": f"Created Zendesk agent profile for {name or employee_id} ({email})"}
-# """Zendesk-like app — customer support ticket management."""
-
-# from typing import Dict, List, Optional
-# from server.apps.base_app import BaseApp
-# from server.schema_drift import SchemaDriftEngine
-
-
-# class ZendeskApp(BaseApp):
-#     APP_NAME = "zendesk"
-
-#     OPERATIONS = [
-#         "get_ticket", "acknowledge_ticket", "set_urgency", "assign_agent",
-#         "escalate_to_jira", "resolve_ticket", "add_note", "list_tickets","create_agent_profile",
-#     ]
-
-#     def __init__(self, drift: SchemaDriftEngine):
-#         super().__init__(drift)
-#         self._records: Dict[str, Dict] = {}
-
-#     # ------------------------------------------------------------------
-#     # BaseApp interface
-#     # ------------------------------------------------------------------
-
-#     def initialize(self, records: List[Dict]) -> None:
-#         self._records = {r["ticket_number"]: r for r in records}
-
-#     def execute(self, operation: str, args: Dict) -> Dict:
-#         method = getattr(self, f"_op_{operation}", None)
-#         if method is None:
-#             return {
-#                 "success": False,
-#                 "message": f"Unknown operation '{operation}'. Available: {', '.join(self.OPERATIONS)}",
-#             }
-#         try:
-#             return method(**args)
-#         except TypeError as exc:
-#             return {"success": False, "message": f"Bad args for '{operation}': {exc}"}
-
-#     def get_state_view(self, max_rows: int = 5) -> str:
-#         open_tickets = [r for r in self._records.values()
-#                         if r.get("state") not in ("resolved", "closed")][:max_rows]
-#         if not open_tickets:
-#             return "No open tickets."
-#         lines = []
-#         for rec in open_tickets:
-#             view = self._to_agent_view(rec)
-#             keep = ["ticket_number", "title",
-#                     "urgency", "priority", "impact_level",
-#                     "agent_email", "handler", "assigned_agent",
-#                     "state", "ticket_state", "resolution_status",
-#                     "customer_id"]
-#             compact = {k: v for k, v in view.items() if k in keep and v is not None}
-#             lines.append(str(compact))
-#         return "\n".join(lines)
-
-#     def count_open_items(self) -> int:
-#         return sum(1 for r in self._records.values()
-#                    if r.get("state") not in ("resolved", "closed"))
-
-#     # ------------------------------------------------------------------
-#     # Workflow completion state checks
-#     # ------------------------------------------------------------------
-
-#     def ticket_acknowledged(self) -> bool:
-#         """True once ZD-001 has been acknowledged (Workflow A step A1)."""
-#         return bool(self._records.get("ZD-001", {}).get("_acknowledged"))
-
-#     def support_queried(self, account_id: str) -> bool:
-#         """True once tickets for account_id were listed (Workflow C step C2)."""
-#         return account_id in self._records.get("ZD-001", {}).get("_queried_accounts", []) or \
-#                any(account_id in r.get("_queried_accounts", []) for r in self._records.values())
-
-#     def profile_created(self) -> bool:
-#         """True once a new agent profile was created (Workflow B step B4)."""
-#         return any(r.get("_profile_created") for r in self._records.values())
-
-#     # ------------------------------------------------------------------
-#     # Operations
-#     # ------------------------------------------------------------------
-
-#     def _op_get_ticket(self, ticket_number: str, customer_id: Optional[str] = None) -> Dict:
-#         # If customer_id provided, look up all tickets for that customer
-#         if customer_id:
-#             matching = [r for r in self._records.values()
-#                         if r.get("customer_id") == customer_id]
-#             # Mark as queried for Workflow C
-#             for r in matching:
-#                 r.setdefault("_queried_accounts", [])
-#                 if customer_id not in r["_queried_accounts"]:
-#                     r["_queried_accounts"].append(customer_id)
-#             if not matching:
-#                 return {"success": True, "data": [],
-#                         "message": f"No tickets found for customer {customer_id}"}
-#             return {
-#                 "success": True,
-#                 "data": [self._to_agent_view(r) for r in matching[:5]],
-#                 "message": f"Found {len(matching)} tickets for {customer_id}",
-#             }
-
-#         rec = self._records.get(ticket_number)
-#         if not rec:
-#             return {"success": False,
-#                     "message": f"Ticket {ticket_number} not found. Use list_tickets to browse."}
-#         rec.setdefault("_queried_accounts", [])
-#         cid = rec.get("customer_id")
-#         if cid and cid not in rec["_queried_accounts"]:
-#             rec["_queried_accounts"].append(cid)
-
-#         return {"success": True, "data": self._to_agent_view(rec),
-#                 "ticket": rec,
-#                 "message": f"Retrieved {ticket_number}"}
-
-#     def _op_acknowledge_ticket(self, ticket_number: str) -> Dict:
-#         rec = self._records.get(ticket_number)
-#         if not rec:
-#             return {"success": False, "message": f"Ticket {ticket_number} not found"}
-#         rec["_acknowledged"] = True
-#         if rec.get("state") == "new":
-#             rec["state"] = "open"
-#         return {"success": True, "ticket": rec,
-#                 "message": f"Acknowledged {ticket_number} — status → open"}
-
-#     def _op_set_urgency(self, ticket_number: str, **kwargs) -> Dict:
-#         schema_error, schema_adapted = self._check_schema_drift(kwargs)
-#         if schema_error:
-#             hint = self._drift.translate_field("urgency", self.APP_NAME)
-#             return {"success": False, "schema_error": schema_error,
-#                     "message": f"Schema error: use '{hint}' not '{schema_error}'"}
-
-#         rec = self._records.get(ticket_number)
-#         if not rec:
-#             return {"success": False, "message": f"Ticket {ticket_number} not found"}
-
-#         new_urgency = (kwargs.get("urgency") or kwargs.get("priority")
-#                        or kwargs.get("impact_level"))
-#         if not new_urgency:
-#             return {"success": False,
-#                     "message": "Provide urgency / priority / impact_level value"}
-
-#         rec["urgency"] = new_urgency
-#         return {"success": True, "schema_adapted": schema_adapted,
-#                 "message": f"{ticket_number} urgency → '{new_urgency}'"}
-
-#     def _op_assign_agent(self, ticket_number: str, **kwargs) -> Dict:
-#         schema_error, schema_adapted = self._check_schema_drift(kwargs)
-#         if schema_error:
-#             hint = self._drift.translate_field("agent_email", self.APP_NAME)
-#             return {"success": False, "schema_error": schema_error,
-#                     "message": f"Schema error: use '{hint}' not '{schema_error}'"}
-
-#         rec = self._records.get(ticket_number)
-#         # For Workflow B profile creation: allow creating a new agent entry
-#         if not rec:
-#             # Create a minimal profile record for the new agent
-#             email = (kwargs.get("agent_email") or kwargs.get("handler")
-#                      or kwargs.get("assigned_agent"))
-#             if not email:
-#                 return {"success": False, "message": f"Ticket {ticket_number} not found"}
-#             # Create a synthetic profile ticket
-#             profile_rec = {
-#                 "ticket_number":    ticket_number,
-#                 "title":            "Agent profile",
-#                 "urgency":          "p3",
-#                 "agent_email":      email,
-#                 "state":            "closed",
-#                 "customer_id":      None,
-#                 "_acknowledged":    False,
-#                 "_queried_accounts": [],
-#                 "_profile_created": True,
-#             }
-#             self._records[ticket_number] = profile_rec
-#             return {"success": True, "schema_adapted": schema_adapted,
-#                     "message": f"Created Zendesk profile for agent '{email}'"}
-
-#         email = (kwargs.get("agent_email") or kwargs.get("handler")
-#                  or kwargs.get("assigned_agent"))
-#         if not email:
-#             return {"success": False,
-#                     "message": "Provide agent_email / handler / assigned_agent value"}
-
-#         rec["agent_email"] = email
-#         rec["_profile_created"] = True
-#         return {"success": True, "schema_adapted": schema_adapted,
-#                 "message": f"{ticket_number} assigned to '{email}'"}
-
-#     def _op_escalate_to_jira(self, ticket_number: str,
-#                               jira_issue_id: Optional[str] = None) -> Dict:
-#         rec = self._records.get(ticket_number)
-#         if not rec:
-#             return {"success": False, "message": f"Ticket {ticket_number} not found"}
-#         rec["state"] = "pending"
-#         rec["escalated_to_jira"] = jira_issue_id or "pending"
-#         return {"success": True,
-#                 "message": f"{ticket_number} escalated to Jira"
-#                            + (f" ({jira_issue_id})" if jira_issue_id else "")}
-
-#     def _op_resolve_ticket(self, ticket_number: str) -> Dict:
-#         rec = self._records.get(ticket_number)
-#         if not rec:
-#             return {"success": False, "message": f"Ticket {ticket_number} not found"}
-#         rec["state"] = "resolved"
-#         return {"success": True, "message": f"{ticket_number} resolved"}
-
-#     def _op_add_note(self, ticket_number: str, note: str) -> Dict:
-#         rec = self._records.get(ticket_number)
-#         if not rec:
-#             return {"success": False, "message": f"Ticket {ticket_number} not found"}
-#         rec.setdefault("notes", []).append(note)
-#         return {"success": True, "message": f"Note added to {ticket_number}"}
-
-#     def _op_list_tickets(self, state: str = "open", customer_id: Optional[str] = None,
-#                          limit: int = 10) -> Dict:
-#         matching = [
-#             r for r in self._records.values()
-#             if (state == "all" or r.get("state") == state)
-#             and (customer_id is None or r.get("customer_id") == customer_id)
-#         ][:limit]
-#         # Mark accounts as queried
-#         if customer_id:
-#             for r in matching:
-#                 r.setdefault("_queried_accounts", [])
-#                 if customer_id not in r["_queried_accounts"]:
-#                     r["_queried_accounts"].append(customer_id)
-
-#         drifted = [self._to_agent_view(r) for r in matching]
-#         keep = ["ticket_number", "title",
-#                 "urgency", "priority", "impact_level",
-#                 "agent_email", "handler", "assigned_agent",
-#                 "state", "ticket_state", "resolution_status",
-#                 "customer_id"]
-#         compact = [{k: v for k, v in r.items() if k in keep and v is not None}
-#                    for r in drifted]
-#         return {
-#             "success": True,
-#             "data": compact,
-#             "message": f"Found {len(compact)} {state} tickets"
-#                        + (f" for {customer_id}" if customer_id else ""),
-#         }
-# def _op_create_agent_profile(self, employee_id: str, email: str,
-#                                  name: Optional[str] = None, **kwargs) -> Dict:
-#         """Create a Zendesk support agent profile (Workflow B step B4)."""
-#         if employee_id in self._records:
-#             return {"success": False,
-#                     "message": f"Profile for {employee_id} already exists"}
-#         profile = {
-#             "ticket_number":     employee_id,
-#             "title":             f"Agent profile — {name or employee_id}",
-#             "urgency":           "p3",
-#             "agent_email":       email,
-#             "state":             "closed",
-#             "customer_id":       None,
-#             "_acknowledged":     False,
-#             "_queried_accounts": [],
-#             "_profile_created":  True,
-#         }
-#         self._records[employee_id] = profile
-#         return {"success": True,
-#                 "message": f"Created Zendesk agent profile for {name or employee_id} ({email})"}
+                "message": f"Created Zendesk agent profile for {name or employee_id} ({email})"}

server/data_generator.py

@@ -226,12 +226,13 @@ def generate_jira_records(n: int = 50, seed: int = SEED) -> List[Dict]:
 
     # Workflow A primary issue: JIRA-001 is unassigned, linked to ACME-001
     records[0].update({
-        "title":          "Customer login fails intermittently",
-        "priority":       "p1",
-        "status":         "open",
-        "customer_id":    "ACME-001",
-        "assignee":       None,
-        "linked_zendesk": None,
+        "title":            "Customer login fails intermittently",
+        "priority":         "p1",
+        "status":           "open",
+        "customer_id":      "ACME-001",
+        "assignee":         None,
+        "linked_zendesk":   None,
+        "_is_primary_bug":  True,   # semantic marker — used by issue_assigned() check
     })
 
     return records
@@ -264,11 +265,12 @@ def generate_zendesk_records(n: int = 40, seed: int = SEED) -> List[Dict]:
 
     # Workflow A primary: ZD-001 is unacknowledged, from ACME-001
     records[0].update({
-        "title":         "Login issue — cannot access my account",
-        "urgency":       "p1",
-        "state":         "new",
-        "customer_id":   "ACME-001",
-        "_acknowledged": False,
+        "title":               "Login issue — cannot access my account",
+        "urgency":             "p1",
+        "state":               "new",
+        "customer_id":         "ACME-001",
+        "_acknowledged":       False,
+        "_is_primary_ticket":  True,   # semantic marker — used by ticket_acknowledged() check
     })
 
     # Workflow C: several tickets from ACME-003
@@ -313,22 +315,24 @@ def generate_salesforce_records(n: int = 30, seed: int = SEED) -> List[Dict]:
 
     # Workflow A: ACME-001 is a paying customer with yellow health
     records[0].update({
-        "company_name": "Acme Corporation",
-        "deal_stage":   "closed_won",
-        "health":       "yellow",
-        "is_paying":    True,
-        "arr":          50_000,
-        "territory":    "west",
+        "company_name":          "Acme Corporation",
+        "deal_stage":            "closed_won",
+        "health":                "yellow",
+        "is_paying":             True,
+        "arr":                   50_000,
+        "territory":             "west",
+        "_is_workflow_a_account": True,   # semantic marker — used by account_checked() check
     })
 
     # Workflow C: ACME-003 is at churn risk
     records[2].update({
-        "company_name": "Globex Systems",
-        "health":       "red",
-        "deal_stage":   "negotiation",
-        "is_paying":    True,
-        "arr":          30_000,
-        "_churn_flagged": False,
+        "company_name":      "Globex Systems",
+        "health":            "red",
+        "deal_stage":        "negotiation",
+        "is_paying":         True,
+        "arr":               30_000,
+        "_churn_flagged":    False,
+        "_is_churn_target":  True,   # semantic marker — used by churn_flagged() / intervention_assigned()
     })
 
     return records
@@ -348,7 +352,9 @@ def generate_workday_records(n: int = 20, seed: int = SEED) -> List[Dict]:
             "name":        f"{random.choice(FIRST_NAMES)} {random.choice(LAST_NAMES)}",
             "level":       random.choice(levels),
             "manager_id":  f"EMP-{random.randint(1, min(i, 5)):03d}" if i > 1 else None,
-            "status":      random.choices(["active", "pending"], weights=[90, 10])[0],
+            # All random employees are "active" — only the new hire below has status="pending"
+            # so list_employees(status="pending") returns exactly one (discoverable) result.
+            "status":      "active",
             "department":  random.choice(departments),
             "territory":   random.choice(territories),
             "email":       f"emp{i}@company.com",
@@ -371,6 +377,7 @@ def generate_workday_records(n: int = 20, seed: int = SEED) -> List[Dict]:
         "_access_provisioned": {},
         "_sla_logged":         False,
         "_onboarding_created": False,
+        "_is_new_hire":        True,   # semantic marker — used by employee_created() check
     })
 
     return records

server/environment.py

@@ -137,12 +137,21 @@ class OrgOSEnvironment:
             self._schema_score = min(1.0, self._schema_score + 0.10)
             self._policy_score = min(1.0, self._policy_score + 0.05)
 
-        # Earn compliance + efficiency for every successful action
-        self._rule_score = min(1.0, self._rule_score + 0.10)
-        self._efficiency = min(1.0, self._efficiency + 0.10)
+        # Earn compliance for every successful compliant action,
+        # normalised so completing all workflow steps earns exactly 1.0.
+        total_steps = max(1, len(self._workflow._steps))
+        earn_rate   = 1.0 / total_steps
+        self._rule_score = min(1.0, self._rule_score + earn_rate)
 
         # 5. Re-evaluate workflow completion
-        self._wf_score = self._workflow.evaluate(self._apps)
+        old_wf_score     = self._wf_score
+        self._wf_score   = self._workflow.evaluate(self._apps)
+        wf_advanced      = self._wf_score > old_wf_score
+
+        # Earn efficiency ONLY when a new workflow step was just completed.
+        # This penalises padding: closing random tickets, repeating ops, etc.
+        if wf_advanced:
+            self._efficiency = min(1.0, self._efficiency + earn_rate)
 
         # 6. SLA check (only if a ticket was touched)
         sla_ok, sla_pen = self._rules.check_sla(

server/workflow_engine.py

@@ -1,7 +1,7 @@
 """Workflow engine — defines and evaluates multi-app workflow completion."""
 
 from dataclasses import dataclass
-from typing import Callable, Dict, List
+from typing import Callable, Dict, List, Optional
 
 
 @dataclass
@@ -14,6 +14,51 @@ class WorkflowStep:
     completion_check: Callable[[Dict], bool]
 
 
+# ---------------------------------------------------------------------------
+# Helpers — look up semantic-marker targets at evaluation time so completion
+# checks are not coupled to specific record IDs in the data generator.
+# ---------------------------------------------------------------------------
+
+def _churn_target_account_id(apps: Dict) -> Optional[str]:
+    """Return the SF account_id flagged as the churn target this episode."""
+    for aid, rec in apps["salesforce"]._records.items():
+        if rec.get("_is_churn_target"):
+            return aid
+    return None
+
+
+def _new_hire_assigned_sf(apps: Dict) -> bool:
+    """Workflow B step B3: the new hire (from Workday) must be the SF owner of an account
+    in their own territory. Threads employee_id + territory across apps automatically."""
+    new_hire = apps["workday"].get_new_hire()
+    if not new_hire:
+        return False
+    return apps["salesforce"].new_hire_assigned_in_territory(
+        new_hire.get("employee_id", ""),
+        new_hire.get("territory", ""),
+    )
+
+
+def _new_hire_assigned_jira(apps: Dict) -> bool:
+    """Workflow B step B4: an open Jira issue must be assigned to the new hire's employee_id."""
+    new_hire = apps["workday"].get_new_hire()
+    if not new_hire:
+        return False
+    return apps["jira"].new_hire_assigned_to_issue(new_hire.get("employee_id", ""))
+
+
+def _support_queried_for_churn_target(apps: Dict) -> bool:
+    """Workflow C step C2: Zendesk must have been queried for the churn target's account."""
+    aid = _churn_target_account_id(apps)
+    return bool(aid) and apps["zendesk"].support_queried(aid)
+
+
+def _bugs_checked_for_churn_target(apps: Dict) -> bool:
+    """Workflow C step C3: Jira list_issues must have been called with customer_id=<churn target>."""
+    aid = _churn_target_account_id(apps)
+    return bool(aid) and apps["jira"].bugs_checked_for(aid)
+
+
 # ---------------------------------------------------------------------------
 # Workflow A: Customer Bug Fix  (Zendesk → Jira → Salesforce → Workday)
 # Agent role: support
@@ -35,7 +80,7 @@ WORKFLOW_A_STEPS = [
         lambda apps: apps["salesforce"].account_checked(),
     ),
     WorkflowStep(
-        "A4", "Assign the pre-existing Jira bug for this customer to an engineer",
+        "A4", "Assign the Jira issue you just created (linked to the Zendesk ticket) to an engineer",
         "jira", "assign_owner",
         lambda apps: apps["jira"].issue_assigned(),
     ),
@@ -52,24 +97,30 @@ WORKFLOW_A_STEPS = [
 # ---------------------------------------------------------------------------
 WORKFLOW_B_STEPS = [
     WorkflowStep(
-        "B1", "Find the pending new hire in Workday and create their onboarding record",
+        "B1",
+        "Find the pending new hire in Workday (list_employees status=pending returns one result) "
+        "and create their onboarding record",
         "workday", "create_onboarding_task",
         lambda apps: apps["workday"].employee_created(),
     ),
     WorkflowStep(
-        "B2", "Provision Jira access for the new employee via Workday",
+        "B2",
+        "Provision Jira access for THAT new hire (use the employee_id from B1)",
         "workday", "provision_access",
         lambda apps: apps["workday"].access_provisioned("jira"),
     ),
     WorkflowStep(
-        "B3", "Assign the new employee to a Salesforce territory account (west region)",
+        "B3",
+        "Assign the new hire as Salesforce account owner for an account in their own territory "
+        "(use employee_id and territory from B1)",
         "salesforce", "assign_account_owner",
-        lambda apps: apps["salesforce"].team_assigned(),
+        _new_hire_assigned_sf,
     ),
     WorkflowStep(
-        "B4", "Create a Zendesk support agent profile for the new employee",
-        "zendesk", "assign_agent",
-        lambda apps: apps["zendesk"].profile_created(),
+        "B4",
+        "Assign an open Jira issue to the new hire (use employee_id from B1 as the assignee)",
+        "jira", "assign_owner",
+        _new_hire_assigned_jira,
     ),
 ]
 
@@ -84,14 +135,16 @@ WORKFLOW_C_STEPS = [
         lambda apps: apps["salesforce"].churn_flagged(),
     ),
     WorkflowStep(
-        "C2", "Query recent support tickets for the at-risk account in Zendesk",
+        "C2", "Query recent support tickets for the at-risk account in Zendesk "
+              "(use the account_id from C1, not a hardcoded value)",
         "zendesk", "get_ticket",
-        lambda apps: apps["zendesk"].support_queried("ACME-003"),
+        _support_queried_for_churn_target,
     ),
     WorkflowStep(
-        "C3", "List open Jira bugs linked to the at-risk account",
+        "C3", "List open Jira bugs for the at-risk account "
+              "(call list_issues with customer_id=<churn account>)",
         "jira", "list_issues",
-        lambda apps: apps["jira"].bugs_checked(),
+        _bugs_checked_for_churn_target,
     ),
     WorkflowStep(
         "C4", "Assign an intervention owner to the at-risk account in Salesforce",
@@ -113,11 +166,13 @@ WORKFLOW_GOALS: Dict[str, str] = {
         "Use list operations to discover relevant record IDs before acting."
     ),
     "B": (
-        "Workflow B — Employee Onboarding: "
-        "A new support engineer has joined the West team and needs to be fully set up. "
-        "Ensure their employment record exists, provision the appropriate tooling access, "
-        "assign them to the correct territory in your CRM, and create their support profile. "
-        "Query the relevant systems to identify the new employee and required accounts."
+        "Workflow B — Manager-Aware Onboarding: "
+        "One employee in Workday is currently pending onboarding (status=pending). "
+        "Find that pending employee, create their onboarding record, provision their Jira access, "
+        "assign them as the Salesforce account owner for an account in THEIR OWN territory, "
+        "and assign them ownership of an open Jira issue. "
+        "Each step's output feeds the next — capture the employee_id and territory from step 1 "
+        "and reuse them in subsequent steps."
     ),
     "C": (
         "Workflow C — Churn Risk Alert: "