perf: harden non-inferable principle implementation #195
+1,520
−108
There are no files selected for viewing
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,185 @@ | ||
| """Org policy quality validation heuristics. | ||
|
|
||
| Applies lightweight checks to detect policies that likely violate the | ||
| non-inferable principle — e.g. policies that describe codebase structure | ||
| (inferable by reading the repo) rather than actionable constraints. | ||
|
|
||
| Examples of **good** policies (non-inferable, actionable): | ||
|
|
||
| - ``"All API responses must include a correlation_id header"`` | ||
| - ``"Never store PII in memory without encryption"`` | ||
| - ``"Escalate budget overruns above $5 to the CFO"`` | ||
|
|
||
| Examples of **bad** policies (inferable or non-actionable): | ||
|
|
||
| - ``"The project uses Python 3.14"`` — discoverable from pyproject.toml | ||
| - ``"src/api/ contains REST controllers"`` — discoverable by reading code | ||
| - ``"x"`` — too short to be actionable | ||
| """ | ||
|
|
||
| import re | ||
| from typing import Final, Literal | ||
|
|
||
| from pydantic import BaseModel, ConfigDict, Field | ||
|
|
||
| from ai_company.observability import get_logger | ||
| from ai_company.observability.events.prompt import PROMPT_POLICY_QUALITY_ISSUE | ||
|
|
||
| logger = get_logger(__name__) | ||
|
|
||
| _MIN_POLICY_LENGTH: Final[int] = 10 | ||
| _MAX_POLICY_LENGTH: Final[int] = 500 | ||
|
|
||
| # Patterns that suggest inferable codebase context rather than a policy. | ||
| _CODE_PATTERNS: Final[tuple[re.Pattern[str], ...]] = ( | ||
| re.compile(r"(?:src|tests|lib|app)/[\w/]+\.py"), # file paths | ||
| re.compile(r"\bfrom\s+\w+\s+import\b"), # Python imports | ||
| re.compile(r"\bimport\s+\w+"), # bare imports | ||
| re.compile(r"\bdef\s+\w+\s*\("), # function definitions | ||
| re.compile(r"\bclass\s+\w+[\s:(]"), # class definitions | ||
| ) | ||
|
Comment on lines
+38
to
+44
There was a problem hiding this comment. The |
||
|
|
||
| # Action verbs that signal an actionable constraint. | ||
| _ACTION_VERBS: Final[frozenset[str]] = frozenset( | ||
| { | ||
| "must", | ||
| "should", | ||
| "always", | ||
| "never", | ||
| "require", | ||
| "ensure", | ||
| "prohibit", | ||
| "enforce", | ||
| "restrict", | ||
| "mandate", | ||
| "avoid", | ||
| "prefer", | ||
| "escalate", | ||
| "approve", | ||
| "deny", | ||
| "reject", | ||
| "validate", | ||
| "verify", | ||
| } | ||
| ) | ||
|
|
||
|
|
||
| class PolicyQualityIssue(BaseModel): | ||
| """A quality issue found in an org policy. | ||
|
|
||
| Attributes: | ||
| policy: The policy text that triggered the issue. | ||
| issue: Human-readable description of the problem. | ||
| severity: ``"warning"`` for advisory, ``"error"`` for likely invalid. | ||
| """ | ||
|
|
||
| model_config = ConfigDict(frozen=True) | ||
|
|
||
| policy: str = Field(description="The policy text that triggered the issue") | ||
| issue: str = Field(description="Human-readable description of the problem") | ||
| severity: Literal["warning", "error"] = Field( | ||
| description="Issue severity (``'error'`` reserved for future stricter checks)", | ||
| ) | ||
|
|
||
|
|
||
| def validate_policy_quality( | ||
| policies: tuple[str, ...], | ||
| ) -> tuple[PolicyQualityIssue, ...]: | ||
| """Check org policies for non-inferable principle violations. | ||
|
|
||
| Applies heuristic checks — results are advisory and never block | ||
| prompt construction. | ||
|
|
||
| Args: | ||
| policies: Org policy texts to validate. | ||
|
|
||
| Returns: | ||
| Tuple of quality issues found (empty if all policies pass). | ||
| """ | ||
| logger.debug( | ||
| PROMPT_POLICY_QUALITY_ISSUE, | ||
| phase="start", | ||
| policy_count=len(policies), | ||
| ) | ||
|
Comment on lines
+107
to
+110
There was a problem hiding this comment.
|
||
| issues: list[PolicyQualityIssue] = [] | ||
| for policy in policies: | ||
| issues.extend(_check_single_policy(policy)) | ||
|
|
||
| for issue in issues: | ||
| logger.warning( | ||
| PROMPT_POLICY_QUALITY_ISSUE, | ||
| policy=issue.policy[:80], | ||
| issue=issue.issue, | ||
| severity=issue.severity, | ||
| ) | ||
|
coderabbitai[bot] marked this conversation as resolved.
coderabbitai[bot] marked this conversation as resolved.
|
||
|
|
||
| return tuple(issues) | ||
|
|
||
|
|
||
| _ACTION_VERB_RE: re.Pattern[str] = re.compile( | ||
| r"\b(?:" + "|".join(_ACTION_VERBS) + r")\b", | ||
| ) | ||
|
Comment on lines
+126
to
+128
There was a problem hiding this comment.
|
||
|
|
||
|
|
||
| def _check_single_policy(policy: str) -> list[PolicyQualityIssue]: | ||
| """Run all heuristic checks on a single policy string. | ||
|
|
||
| Args: | ||
| policy: The policy text to validate. | ||
|
|
||
| Returns: | ||
| List of quality issues found (empty if the policy passes all checks). | ||
| """ | ||
| found: list[PolicyQualityIssue] = [] | ||
|
|
||
| if len(policy) < _MIN_POLICY_LENGTH: | ||
| found.append( | ||
| PolicyQualityIssue( | ||
| policy=policy, | ||
| issue=( | ||
| f"Too short ({len(policy)} chars) — likely not an actionable policy" | ||
| ), | ||
| severity="warning", | ||
| ), | ||
| ) | ||
|
|
||
| if len(policy) > _MAX_POLICY_LENGTH: | ||
| found.append( | ||
| PolicyQualityIssue( | ||
| policy=policy, | ||
| issue=( | ||
| f"Too long ({len(policy)} chars) — " | ||
| f"may contain inferable context rather than a policy" | ||
| ), | ||
| severity="warning", | ||
| ), | ||
| ) | ||
|
|
||
| for pattern in _CODE_PATTERNS: | ||
| if pattern.search(policy): | ||
| found.append( | ||
| PolicyQualityIssue( | ||
| policy=policy, | ||
| issue=( | ||
| "Contains code patterns (file paths, imports, or " | ||
| "definitions) — likely inferable from the codebase" | ||
| ), | ||
| severity="warning", | ||
| ), | ||
| ) | ||
| break # One code-pattern match is sufficient. | ||
|
|
||
| policy_lower = policy.lower() | ||
| if not _ACTION_VERB_RE.search(policy_lower): | ||
| found.append( | ||
| PolicyQualityIssue( | ||
| policy=policy, | ||
| issue=( | ||
| "Missing action verbs (must, should, always, never, " | ||
| "etc.) — may not be an actionable policy" | ||
| ), | ||
| severity="warning", | ||
| ), | ||
| ) | ||
|
|
||
| return found | ||
|
Contributor
There was a problem hiding this comment. 🛠️ Refactor suggestion | 🟠 Major Split This helper is already over the 50-line limit and now mixes length, code-pattern, and action-verb heuristics. Extract those checks into small helpers so future rule changes stay isolated and easier to test. Refactor sketch def _check_single_policy(policy: str) -> list[PolicyQualityIssue]:
- found: list[PolicyQualityIssue] = []
-
- if len(policy) < _MIN_POLICY_LENGTH:
- found.append(...)
-
- if len(policy) > _MAX_POLICY_LENGTH:
- found.append(...)
-
- for pattern in _CODE_PATTERNS:
- if pattern.search(policy):
- found.append(...)
- break
-
- policy_lower = policy.lower()
- if not _ACTION_VERB_RE.search(policy_lower):
- found.append(...)
-
- return found
+ return [
+ *_check_policy_length(policy),
+ *_check_code_patterns(policy),
+ *_check_action_verbs(policy),
+ ]As per coding guidelines "Keep functions under 50 lines and files under 800 lines". 🤖 Prompt for AI Agents |
||
Oops, something went wrong.
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
prompt_tokensis populated fromresult.system_prompt.estimated_tokens, but the system prompt message is included inctx.conversationand is resent on every provider call. Sincetokens_per_taskaggregates tokens across all turns,prompt_token_ratiowill be underestimated for multi-turn runs. Consider either (a) makingprompt_tokensrepresent total prompt tokens across the run (e.g., estimate ×result.total_turns), or (b) renaming the field to clarify it's per-call and adjusting the ratio/warning accordingly.