Skip to content
Draft
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
19 changes: 19 additions & 0 deletions handoff.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,19 @@
# Rocq wired review handoff

## Stack

- Base: `contrib/rocq-backend`.
- This branch removes `CoqInterface` and runs the agent, CLI, interactive session, context search, rollback, and proof saving through `ProverBackend` and `CoqPytBackend`.
- The backend factory exposes only `rocq`; Lean and Isabelle remain in their own review stacks.

## Verification

- `PYTHONPATH=proof-search pytest -q -m 'not integration and not live_api' proof-search/tests` — 92 passed, 37 deselected, one pre-existing `PytestReturnNotNoneWarning`.
- `PYTHONPATH=proof-search pytest -q proof-search/tests/test_agent_rocq_workflow.py` — 1 passed against the installed Rocq/coq-lsp toolchain in 7.27s.
- `rg -n "backend\\.coq_interface|CoqInterface|coq_interface" proof-search --glob '*.py'` — no matches.

## Next task

Review the narrow fork stack [#1](https://github.com/dingf3ng/LemmaNet/pull/1) → [#2](https://github.com/dingf3ng/LemmaNet/pull/2) → [#5](https://github.com/dingf3ng/LemmaNet/pull/5). The corresponding upstream sequence is [#3](https://github.com/NUS-Program-Verification/LemmaNet/pull/3) → [#4](https://github.com/NUS-Program-Verification/LemmaNet/pull/4) → draft [#7](https://github.com/NUS-Program-Verification/LemmaNet/pull/7).

After upstream #4 merges, rebase this branch onto the updated upstream `main`, push with `--force-with-lease`, and mark #7 ready for review.
438 changes: 315 additions & 123 deletions proof-search/agent/context_manager.py

Large diffs are not rendered by default.

306 changes: 75 additions & 231 deletions proof-search/agent/context_search.py
Original file line number Diff line number Diff line change
Expand Up @@ -304,256 +304,100 @@ def _categorize_entries(self, entries: List[Dict[str, str]]) -> Dict[str, int]:
return categories


class CoqCommandSearch:
"""Handles Coq command-line search operations with adaptive result reduction."""

def __init__(self, coq_interface):
"""
Initialize with CoqInterface from backend.coq_interface.

Args:
coq_interface: Instance of CoqInterface from backend.coq_interface
"""
self.coq = coq_interface

class ProverCommandSearch:
"""Run native context queries through a typed prover backend."""

def __init__(self, backend):
from backend.prover_backend import ProverBackend

if not isinstance(backend, ProverBackend):
raise TypeError("context search requires a ProverBackend")
self.backend = backend
self.reducer = ResultReducer()
self.logger = setup_logger("ProverCommandSearch")

# Setup logger
self.logger = setup_logger("CoqCommandSearch")

# Ensure the CoqInterface is loaded
if not hasattr(self.coq, 'proof_file') or self.coq.proof_file is None:
try:
self.coq.load()
self.logger.info("✅ CoqInterface loaded successfully")
except Exception as e:
self.logger.error(f"❌ Failed to load CoqInterface: {e}")

def _create_search_result(self, content: str, query: str, query_type: str, goal_context: str = "") -> SearchResult:
"""Create a SearchResult with adaptive size reduction."""
def _create_search_result(
self,
content: str,
query: str,
query_type: str,
goal_context: str = "",
) -> SearchResult:
original_size = len(content) if content else 0

# Apply adaptive reduction
reduced_content, reduction_method = self.reducer.reduce_result(content, query_type, goal_context)
reduced_content, reduction_method = self.reducer.reduce_result(
content, query_type, goal_context
)
final_size = len(reduced_content) if reduced_content else 0

# Log reduction if applied
if reduction_method != "none":
self.logger.debug(f"Applied {reduction_method}: {original_size} → {final_size} chars ({original_size - final_size} saved)")

return SearchResult(
content=reduced_content,
source='coq_command',
relevance_score=1.0 if reduced_content and "No results found" not in reduced_content else 0.0,
source="prover_query",
relevance_score=(
1.0
if reduced_content and "No results found" not in reduced_content
else 0.0
),
metadata={
'query': query,
'type': query_type,
'reduction_applied': reduction_method,
'original_size': original_size,
'size_saved': original_size - final_size
"query": query,
"type": query_type,
"reduction_applied": reduction_method,
"original_size": original_size,
"size_saved": original_size - final_size,
},
result_size=final_size,
original_size=original_size,
reduction_applied=reduction_method
reduction_applied=reduction_method,
)

def search_lemma(self, lemma_name: str, goal_context: str = "") -> SearchResult:
"""Search for a specific lemma or theorem."""
query = f"Search {lemma_name}."
result = self.coq.search(query)
return self._create_search_result(result, query, 'search_lemma', goal_context)

def search_pattern(self, pattern: str, goal_context: str = "") -> SearchResult:
"""Search for theorems matching a pattern."""
# Clean the pattern for Coq search
if not pattern.startswith('(') and not pattern.endswith(')'):
pattern = f"({pattern})"

query = f"Search {pattern}."
result = self.coq.search(query)
return self._create_search_result(result, query, 'search_pattern', goal_context)

def print_definition(self, identifier: str) -> SearchResult:
"""Print the definition of an identifier."""
query = f"Print {identifier}."
result = self.coq.search(query)
return self._create_search_result(result, query, 'print_definition')

def print_assumptions(self, identifier: str = None) -> SearchResult:
"""Print assumptions of an identifier or all assumptions."""
if identifier:
query = f"Print Assumptions {identifier}."
else:
query = "Print Assumptions."

result = self.coq.search(query)
return self._create_search_result(result, query, 'print_assumptions')

def locate_definition(self, identifier: str) -> SearchResult:
"""Locate the definition of an identifier."""
query = f"Locate {identifier}."
result = self.coq.search(query)
return self._create_search_result(result, query, 'locate_definition')

def about_identifier(self, identifier: str) -> SearchResult:
"""Get information about an identifier."""
query = f"About {identifier}."
result = self.coq.search(query)
return self._create_search_result(result, query, 'about_identifier')

def check_term(self, term: str) -> SearchResult:
"""Check the type of a term."""
query = f"Check {term}."
result = self.coq.search(query)
return self._create_search_result(result, query, 'check_term')

def auto_search(self, search_request: str, goal_context: str = "") -> SearchResult:
"""Automatically determine search type and execute with adaptive reduction."""
search_request = search_request.strip()

# All commands now go through the enhanced search() method
result = self.coq.search(search_request)

# Determine type from command
cmd_type = search_request.split()[0].lower() if search_request else 'unknown'
type_mapping = {
'search': 'direct_search',
'print': 'direct_print',
'locate': 'direct_locate',
'about': 'direct_about',
'check': 'direct_check'
}

query_type = type_mapping.get(cmd_type, 'auto_search')
return self._create_search_result(result, search_request, query_type, goal_context)

def execute_coq_query(self, query_type: str, identifier: str = None, pattern: str = None, goal_context: str = "") -> SearchResult:
"""Execute a Coq query by type with parameters and adaptive reduction."""
async def _query(
self, command: str, query_type: str, goal_context: str = ""
) -> SearchResult:
from backend.prover_backend import CommandRejectedError

try:
if query_type.lower() == 'search':
if pattern:
return self.search_pattern(pattern, goal_context)
elif identifier:
return self.search_lemma(identifier, goal_context)
else:
error_msg = "Search requires either identifier or pattern"
return SearchResult(
content=error_msg,
source='coq_command',
relevance_score=0.0,
metadata={'query_type': query_type, 'error': 'Missing parameters'},
result_size=len(error_msg)
)
elif query_type.lower() == 'print':
if identifier:
return self.print_definition(identifier)
else:
error_msg = "Print requires identifier"
return SearchResult(
content=error_msg,
source='coq_command',
relevance_score=0.0,
metadata={'query_type': query_type, 'error': 'Missing identifier'},
result_size=len(error_msg)
)
elif query_type.lower() == 'print_assumptions':
return self.print_assumptions(identifier)
elif query_type.lower() == 'locate':
if identifier:
return self.locate_definition(identifier)
else:
error_msg = "Locate requires identifier"
return SearchResult(
content=error_msg,
source='coq_command',
relevance_score=0.0,
metadata={'query_type': query_type, 'error': 'Missing identifier'},
result_size=len(error_msg)
)
elif query_type.lower() == 'about':
if identifier:
return self.about_identifier(identifier)
else:
error_msg = "About requires identifier"
return SearchResult(
content=error_msg,
source='coq_command',
relevance_score=0.0,
metadata={'query_type': query_type, 'error': 'Missing identifier'},
result_size=len(error_msg)
)
elif query_type.lower() == 'check':
if identifier:
return self.check_term(identifier)
else:
error_msg = "Check requires term"
return SearchResult(
content=error_msg,
source='coq_command',
relevance_score=0.0,
metadata={'query_type': query_type, 'error': 'Missing term'},
result_size=len(error_msg)
)
else:
error_msg = f"Unknown query type: {query_type}"
return SearchResult(
content=error_msg,
source='coq_command',
relevance_score=0.0,
metadata={'query_type': query_type, 'error': 'Unknown query type'},
result_size=len(error_msg)
)
except Exception as e:
error_msg = f"Error executing {query_type}: {str(e)}"
return SearchResult(
content=error_msg,
source='coq_command',
relevance_score=0.0,
metadata={'query_type': query_type, 'error': str(e)},
result_size=len(error_msg)
result = await self.backend.query(command)
except CommandRejectedError as error:
message = error.feedback[0].message if error.feedback else str(error)
return self._create_search_result(message, command, query_type, goal_context)
return self._create_search_result(
result.output, result.command, query_type, goal_context
)

async def auto_search(
self, search_request: str, goal_context: str = ""
) -> SearchResult:
command = search_request.strip()
if not command:
return self._create_search_result(
"No results found.", command, "direct_search", goal_context
)
query_type = command.split(maxsplit=1)[0].lower()
if query_type == "search":
query_type = "direct_search"
return await self._query(command, query_type, goal_context)



class ContextSearch:
"""
Simplified context search interface with adaptive result reduction.
"""

def __init__(self, coq_interface, history_file: str = None):
"""
Initialize context search with CoqInterface.

Args:
coq_interface: Instance of CoqInterface from backend.coq_interface
history_file: Ignored (kept for backward compatibility)
"""
self.coq_search = CoqCommandSearch(coq_interface)
"""Context-query policy with adaptive result reduction."""

def __init__(self, backend, history_file: str | None = None):
del history_file
self.command_search = ProverCommandSearch(backend)
self.logger = setup_logger("ContextSearch")

def search(self, query: str, goal_context: str = "") -> SearchResult:
"""
Simplified search interface with adaptive result reduction.

Args:
query: Search query string
goal_context: Current proof goal context for relevance ranking

Returns:
SearchResult from Coq command execution
"""

async def search(
self, query: str, goal_context: str = ""
) -> SearchResult:
try:
return self.coq_search.auto_search(query, goal_context)
except Exception as e:
self.logger.error(f"Error in Coq command search: {e}")
error_message = f"Search error: {str(e)}"
return await self.command_search.auto_search(query, goal_context)
except Exception as error:
self.logger.error(f"Error in prover command search: {error}")
message = f"Search error: {error}"
return SearchResult(
content=error_message,
source='coq_command',
content=message,
source="prover_query",
relevance_score=0.0,
metadata={'query': query, 'error': str(e)},
result_size=len(error_message)
metadata={"query": query, "error": str(error)},
result_size=len(message),
)

def execute_coq_query(self, query_type: str, identifier: str = None, pattern: str = None, goal_context: str = "") -> SearchResult:
"""Execute a Coq query with adaptive result reduction."""
return self.coq_search.execute_coq_query(query_type, identifier, pattern, goal_context)

Loading